如何以阻塞方式有效地提取 futures::Stream 的第一个元素？答案

【问题标题】：How can I efficiently extract the first element of a futures::Stream in a blocking manner?如何以阻塞方式有效地提取 futures::Stream 的第一个元素？
【发布时间】：2019-03-02 22:55:35
【问题描述】：

我有以下方法：

pub fn load_names(&self, req: &super::MagicQueryType) -> ::grpcio::Result<::grpcio::ClientSStreamReceiver<String>> {

我的目标是获得grpcio::ClientSStreamReceiver 的第一个元素；我不在乎其他名字：

let name: String = load_names(query)?.wait().nth(0)?;

在nth(0) 之前调用wait() 似乎效率低下，因为我相信wait() 会阻塞流，直到它接收到所有元素。

如何在不触发构建错误的情况下编写更高效的解决方案（即nth(0).wait()）？ Rust 对 futures::stream::Stream 的构建错误让我非常困惑。

Rust playground 不支持grpcio = "0.4.4"，所以我无法提供链接。

【问题讨论】：

Playground 没有板条箱并不意味着您不应该提供 MCVE。

标签： stream rust

【解决方案1】：

要以阻塞方式提取futures::Stream 的第一个元素，您应该通过调用executor::block_on_stream 将Stream 转换为迭代器，然后调用Iterator::next。

use futures::{executor, stream, Stream}; // 0.3.4
use std::iter;

fn example() -> impl Stream<Item = i32> {
    stream::iter(iter::repeat(42))
}

fn main() {
    let v = executor::block_on_stream(example()).next();
    println!("{:?}", v);
}

如果您使用的是 Tokio，您可以使用 StreamExt::into_future 将 Stream 转换为 Future 并使用 #[tokio::main] 注释函数：

use futures::{stream, Stream, StreamExt}; // 0.3.4
use std::iter;
use tokio; // 0.2.13

fn example() -> impl Stream<Item = i32> {
    stream::iter(iter::repeat(42))
}

#[tokio::main]
async fn just_one() -> Option<i32> {
    let (i, _stream) = example().into_future().await;
    i
}

fn main() {
    println!("{:?}", just_one());
}

另见：

【讨论】：