4

I have an asynchronous Stream and I'd like to get the first value out of it. How can I do so?

use futures::Stream; // 0.3.5

async fn example<T>(s: impl Stream<Item = T>) -> Option<T> {
    todo!("What goes here?")
}
Shepmaster
  • 388,571
  • 95
  • 1,107
  • 1,366
  • Note that this question is because I had an answer for another post that was poorly titled and edited, and ultimately the answer was irrelevant. Rather than just delete it, I'm making it into a self-answered question. – Shepmaster Sep 09 '20 at 02:37

1 Answers1

5

You can use StreamExt::next:

use futures::{Stream, StreamExt}; // 0.3.5

async fn example<T>(mut s: impl Stream<Item = T> + Unpin) -> Option<T> {
    s.next().await
}

You can use StreamExt::into_future:

use futures::{FutureExt, Stream, StreamExt}; // 0.3.5

async fn example<T>(s: impl Stream<Item = T> + Unpin) -> Option<T> {
    s.into_future().map(|(v, _)| v).await
}

In rare cases, you may want to use future::poll_fn to have complete control:

use futures::{future, task::Poll, Stream, StreamExt}; // 0.3.5

async fn example<T>(mut s: impl Stream<Item = T> + Unpin) -> Option<T> {
    future::poll_fn(|ctx| {
        // Could use methods like `Poll::map` or
        // the `?` operator instead of a `match`.
        match s.poll_next_unpin(ctx) {
            Poll::Ready(v) => {
                // Do any special logic here
                Poll::Ready(v)
            }
            Poll::Pending => Poll::Pending,
        }
    })
    .await
}

See also:

More broadly

If you wanted to act upon all the values in the stream, producing a single value, you can use StreamExt::fold:

use futures::{Stream, StreamExt}; // 0.3.5

async fn example(s: impl Stream + Unpin) -> usize {
    s.fold(0, |st, _| async move { st + 1 }).await
}

If you wanted to act upon all the values in the stream without producing a value, you can use StreamExt::for_each:

use futures::{Stream, StreamExt}; // 0.3.5

async fn example<I: std::fmt::Debug>(s: impl Stream<Item = I> + Unpin) {
    s.for_each(|i| async {
        dbg!(i);
    })
    .await;
}

See also:

Unpin

These example all require that the incoming Stream implement Unpin. You could also pin the stream yourself via Box::pin or the pin_mut! macro.

See also:

Shepmaster
  • 388,571
  • 95
  • 1,107
  • 1,366