在调用异步 fns 时创建值流？答案

【问题标题】：Creating a stream of values while calling async fns?在调用异步 fns 时创建值流？
【发布时间】：2019-06-22 23:51:38
【问题描述】：

我不知道如何提供一个Stream await 异步函数来获取流值所需的数据。

我尝试直接实现 Stream 特征，但我遇到了问题，因为我想使用像 awaiting 这样的异步东西，编译器不希望我调用异步函数。

我认为我缺少一些关于 Stream 目标的背景知识，我只是错误地攻击了这一点，也许我根本不应该看 Stream，但我不知道还有哪里可以转。我见过其他可能有用的functions in the stream module，但我不确定如何存储任何状态并使用这些功能。

作为我实际目标的略微简化版本，我想提供来自 AsyncRead 对象（即 TCP 流）的 64 字节 Vecs 流，但还要在任何逻辑结束时存储一些状态为流生成值，在本例中为计数器。

pub struct Receiver<T>
where
    T: AsyncRead + Unpin,
{
    readme: T,
    num: u64,
}

// ..code for a simple `new() -> Self` function..

impl<T> Stream for Receiver<T>
where
    T: AsyncRead + Unpin,
{
    type Item = Result<Vec<u8>, io::Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let mut buf: [u8; 64] = [0; 64];
        match self.readme.read_exact(&mut buf).await {
            Ok(()) => {
                self.num += 1;
                Poll::Ready(Some(Ok(buf.to_vec())))
            }
            Err(e) => Poll::Ready(Some(Err(e))),
        }
    }
}

这无法构建，说

error[E0728]: `await` is only allowed inside `async` functions and blocks

我正在使用rustc 1.36.0-nightly (d35181ad8 2019-05-20)，而我的Cargo.toml 看起来像这样：

[dependencies]
futures-preview = { version = "0.3.0-alpha.16", features = ["compat", "io-compat"] }
pin-utils = "0.1.0-alpha.4"

【问题讨论】：

标签： rust async-await stream

【解决方案1】：

用户Matthias247从reddit post复制/粘贴的答案：

不幸的是，目前不可能 - Streams 必须手动实现，并且不能使用 async fn。未来是否有可能改变这一点尚不清楚。

您可以通过定义一个使用 Future 的不同 Stream 特征来解决它，例如：
trait Stream<T> { 
   type NextFuture: Future<Output=T>;

   fn next(&mut self) -> Self::NextFuture; 
}
This article 和 this futures-rs issue 有更多相关信息。

【讨论】：

【解决方案2】：

您可以使用gen-stream crate：

#![feature(generators, generator_trait, gen_future)]

use {
    futures::prelude::*,
    gen_stream::{gen_await, GenTryStream},
    pin_utils::unsafe_pinned,
    std::{
        io,
        marker::PhantomData,
        pin::Pin,
        sync::{
            atomic::{AtomicU64, Ordering},
            Arc,
        },
        task::{Context, Poll},
    },
};

pub type Inner = Pin<Box<dyn Stream<Item = Result<Vec<u8>, io::Error>> + Send>>;

pub struct Receiver<T> {
    inner: Inner,
    pub num: Arc<AtomicU64>,
    _marker: PhantomData<T>,
}

impl<T> Receiver<T> {
    unsafe_pinned!(inner: Inner);
}

impl<T> From<T> for Receiver<T>
where
    T: AsyncRead + Unpin + Send + 'static,
{
    fn from(mut readme: T) -> Self {
        let num = Arc::new(AtomicU64::new(0));

        Self {
            inner: Box::pin(GenTryStream::from({
                let num = num.clone();
                static move || loop {
                    let mut buf: [u8; 64] = [0; 64];
                    match gen_await!(readme.read_exact(&mut buf)) {
                        Ok(()) => {
                            num.fetch_add(1, Ordering::Relaxed);
                            yield Poll::Ready(buf.to_vec())
                        }
                        Err(e) => return Err(e),
                    }
                }
            })),
            num,
            _marker: PhantomData,
        }
    }
}

impl<T> Stream for Receiver<T>
where
    T: AsyncRead + Unpin,
{
    type Item = Result<Vec<u8>, io::Error>;

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

【讨论】：