How can I efficiently extract the first element of

Question

I've got the following method:

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

My goal is to get the very first element of grpcio::ClientSStreamReceiver; I don't care about the other names:

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

It seems inefficient to call wait() before nth(0) as I believe wait() blocks the stream until it receives all the elements.

How can I write a more efficient solution (i.e., nth(0).wait()) without triggering build errors? Rust's build errors for futures::stream::Stream look extremely confusing to me.

The Rust playground doesn't support grpcio = "0.4.4" so I cannot provide a link.

Answer 1

To extract the first element of a futures::Stream in a blocking manner, you should convert the Stream to an iterator by calling Stream::wait and then call Iterator::next. Reading the documentation for Stream::wait shows that it's basically the most efficient implementation:

This iterator will block the current thread on each call to next if the item in the stream isn't ready yet.

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

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

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

If you are using Tokio, you can convert the Stream into a Future and call Runtime::block_on:

use std::iter;
use tokio::prelude::*; // 0.1.15

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

fn main() {
    let mut runtime = tokio::runtime::Runtime::new().expect("Unable to create a runtime");
    let r = runtime.block_on(example().into_future());
    if let Ok((v, _)) = r {
        println!("{:?}", v);
    }
}

See also:

• How do I synchronously return a value calculated in an asynchronous Future in stable Rust?
• How to select between a future and stream in Rust?