Stream

大家有没有想过， Rust 中的迭代器在迭代时能否异步进行？若不可以，是不是有相应的解决方案？

以上的问题其实很重要，因为在实际场景中，迭代一个集合，然后异步的去执行是很常见的需求，好在 Tokio 为我们提供了 stream，我们可以在异步函数中对其进行迭代，甚至和迭代器 Iterator 一样，stream 还能使用适配器，例如 map ! Tokio 在 StreamExt 特征上定义了常用的适配器。

要使用 stream ，目前还需要手动引入对应的包：

#![allow(unused)]
fn main() {
tokio-stream = "0.1"
}

stream 没有放在 tokio 包的原因在于标准库中的 Stream 特征还没有稳定，一旦稳定后，stream 将移动到 tokio 中来

实现 Stream 特征

如果大家还没忘记 Future 特征，那 Stream 特征相信你也会很快记住，因为它们非常类似：

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Stream {
    type Item;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;

    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, None)
    }
}
}

Stream::poll_next() 函数跟 Future::poll 很相似，区别就是前者为了从 stream 收到多个值需要重复的进行调用。就像在 深入async 章节提到的那样，当一个 stream 没有做好返回一个值的准备时，它将返回一个 Poll::Pending ，同时将任务的 waker 进行注册。一旦 stream 准备好后， waker 将被调用。

通常来说，如果想要手动实现一个 Stream，需要组合 Future 和其它 Stream。下面，还记得在深入async 中构建的 Delay Future 吗？现在让我们来更进一步，将它转换成一个 stream，每 10 毫秒生成一个值，总共生成 3 次:

#![allow(unused)]
fn main() {
use tokio_stream::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

struct Interval {
    rem: usize,
    delay: Delay,
}

impl Stream for Interval {
    type Item = ();

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<()>>
    {
        if self.rem == 0 {
            // 去除计时器实现
            return Poll::Ready(None);
        }

        match Pin::new(&mut self.delay).poll(cx) {
            Poll::Ready(_) => {
                let when = self.delay.when + Duration::from_millis(10);
                self.delay = Delay { when };
                self.rem -= 1;
                Poll::Ready(Some(()))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}
}

async-stream

手动实现 Stream 特征实际上是相当麻烦的事，不幸地是，Rust 语言的 async/await 语法目前还不能用于定义 stream，虽然相关的工作已经在进行中。

作为替代方案，async-stream 包提供了一个 stream! 宏，它可以将一个输入转换成 stream，使用这个包，上面的代码可以这样实现：

#![allow(unused)]
fn main() {
use async_stream::stream;
use std::time::{Duration, Instant};

stream! {
    let mut when = Instant::now();
    for _ in 0..3 {
        let delay = Delay { when };
        delay.await;
        yield ();
        when += Duration::from_millis(10);
    }
}
}

嗯，看上去还是相当不错的，代码可读性大幅提升！

是不是发现了一个关键字 yield ，他是用来配合生成器使用的。详见原文

Rust语言圣经(Rust Course)

Stream

迭代

mini-redis 广播

适配器

实现 Stream 特征

async-stream