Streams in Rust With Futures and Tokio Stream Explained

A Future produces one value later. A Stream produces many values over time: lines from a socket, rows from a query, events from a queue. It is the async cousin of Iterator, with poll_next where the iterator has next, and the same family of adapters (map, filter, take, fold) as methods on an extension trait. The trait itself is not in the standard library yet, so streams live in the futures and tokio-stream crates, and the parts you must know are small: how to consume one, how to build one, how to run work concurrently over one, and how to end one. Everything compiled on Rust 1.99, edition 2024, with tokio 1, tokio-stream 0.1 and futures 0.3.

The trait, and what is not in std

pub trait Stream {
    type Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}

poll_next returns Poll::Pending when no item is ready (and arranges a wakeup), Poll::Ready(Some(item)) for an item, and Poll::Ready(None) when the stream has ended. After it returns None, calling it again is allowed to do anything, usually panic or return None forever. The trait is defined in the futures-core crate and re-exported by both futures and tokio-stream; the standard library has an unstable AsyncIterator and has not stabilised an equivalent. Practically, you import futures::StreamExt or tokio_stream::StreamExt to get the adapter methods, and the two are interchangeable for most work.

Consuming a stream

The idiom is while let Some(item) = stream.next().await. Because the stream must be pinned to be polled, a stream stored in a local usually needs pin! unless it is Unpin (most of the ones from these crates are):

use futures::stream::{self, StreamExt};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut numbers = stream::iter(1..=10);

    let mut evens_squared = Vec::new();
    while let Some(n) = numbers.next().await {
        if n % 2 == 0 {
            evens_squared.push(n * n);
        }
    }
    println!("{evens_squared:?}");

    let total: u32 = stream::iter(1..=10).filter(|n| futures::future::ready(n % 2 == 1)).map(|n| n * 10).fold(0, |acc, n| async move { acc + n }).await;
    println!("{total}");

    let first_three: Vec<u32> = stream::iter(100..).take(3).collect().await;
    println!("{first_three:?}");
}

[4, 16, 36, 64, 100]
250
[100, 101, 102]

The closures in filter and fold return futures, not plain values, which is what makes them differ from the iterator versions: filter(|n| ready(...)) because the predicate may itself await, and fold(..., |acc, n| async move { ... }). The cost is a little ceremony. For predicates that never await, futures::future::ready is the shortest wrapper; for the common case there are sync-closure variants in other crates, and then versus map is the same choice (async closure versus plain).

Building a stream

You rarely implement poll_next by hand. Four constructors cover almost everything:

The unfold constructor is the workhorse for paginated APIs: the state is the next page token, each step fetches a page, yields its rows and returns the following token, and the stream ends when the token is gone.

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

async fn fetch_page(token: u32) -> (Vec<String>, Option<u32>) {
    sleep(Duration::from_millis(50)).await;
    let rows: Vec<String> = (0..2).map(|i| format!("row-{}", token * 2 + i)).collect();
    let next = if token < 2 { Some(token + 1) } else { None };
    (rows, next)
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let pages = stream::unfold(Some(0u32), |state| async move {
        let token = state?;
        let (rows, next) = fetch_page(token).await;
        Some((stream::iter(rows), next))
    });

    let rows: Vec<String> = pages.flatten().collect().await;
    println!("{rows:?}");
}

["row-0", "row-1", "row-2", "row-3", "row-4", "row-5"]

Pages are fetched lazily: the second request is not made until the consumer asks for items beyond the first page. If the consumer stops early with take(3), only two pages are ever requested. That laziness is the practical argument for a stream over "fetch all pages into a Vec": memory stays bounded and the work follows demand.

Channels and timers as streams

Wrapping a channel receiver gives you the adapters for free, so that "merge two event sources, drop duplicates, stop after a deadline" is a short chain:

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx_a, rx_a) = mpsc::channel::<&str>(4);
    let (tx_b, rx_b) = mpsc::channel::<&str>(4);

    tokio::spawn(async move {
        for (ms, msg) in [(10, "a1"), (30, "a2")] {
            sleep(Duration::from_millis(ms)).await;
            tx_a.send(msg).await.unwrap();
        }
    });
    tokio::spawn(async move {
        sleep(Duration::from_millis(20)).await;
        tx_b.send("b1").await.unwrap();
    });

    let merged = ReceiverStream::new(rx_a).merge(ReceiverStream::new(rx_b));
    let events = merged.timeout(Duration::from_millis(100));
    tokio::pin!(events);

    while let Some(item) = events.next().await {
        match item {
            Ok(msg) => println!("event {msg}"),
            Err(_) => {
                println!("quiet for 100 ms, stopping");
                break;
            }
        }
    }
}

event a1
event b1
event a2

The tokio::pin!(events) line is not decoration. The timeout adapter embeds a timer that cannot be moved once polled, so the stream is not Unpin, and calling .next() on it without pinning fails with "PhantomPinned cannot be unpinned" followed by the suggestion "consider using the pin! macro". Streams from stream::iter, channel wrappers and most simple adapters are Unpin and need nothing; whenever an adapter wraps a timer or an async block, expect to pin. Box::pin(stream) works too, when the stream must be stored or returned. merge interleaves the two sources by readiness, timeout turns silence into an item (Err(Elapsed)) that applies to each gap between items, and the loop stops on it. The same code with threads would need a select loop and careful bookkeeping. Note the cancel-safety angle: next() on a stream is cancel safe when the underlying stream is, which holds for these wrappers, so using it in a select! loop is fine. The caveat is in cancellation safety.

Writing your own adapter

When you do need a custom stream, the smallest honest implementation wraps another one. This one yields only every second item, and it shows the three things that make a hand-written stream correct: delegate poll_next to the inner stream, preserve Pending (the inner stream has already registered the waker), and end when the inner stream ends:

use futures::stream::{self, Stream, StreamExt};
use std::pin::Pin;
use std::task::{Context, Poll};

struct EverySecond<S> {
    inner: S,
    skip_next: bool,
}

impl<S: Stream + Unpin> Stream for EverySecond<S> {
    type Item = S::Item;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
        loop {
            match Pin::new(&mut self.inner).poll_next(cx) {
                Poll::Ready(Some(item)) => {
                    self.skip_next = !self.skip_next;
                    if !self.skip_next {
                        return Poll::Ready(Some(item));
                    }
                }
                other => return other,
            }
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let s = EverySecond { inner: stream::iter(1..=9), skip_next: false };
    println!("{:?}", s.collect::<Vec<_>>().await);
}

[2, 4, 6, 8]

The loop continues when an item is skipped, because returning Pending without a registered waker would stall the stream. That is the rule from the future side too: only return Pending if someone below you did. For wrapping non-Unpin streams you would use pin-project-lite instead of the Unpin bound; most code is better off composing existing adapters.

Concurrency over a stream

The most useful adapters run work concurrently. buffer_unordered(n) polls up to n futures from the stream at once and yields results as they complete, and buffered(n) does the same but keeps input order. They turn "for each item, do an async call" into a bounded-concurrency pipeline in one line, which is the subject of buffer_unordered.

Ending and error handling

A stream ends with None. A stream of Results keeps going after an error unless you stop it: try_next() (from TryStreamExt) returns Result<Option<T>, E> so the first error can be propagated with ?, and try_collect() short-circuits likewise:

use futures::stream::{self, TryStreamExt};

async fn run(inputs: Vec<&str>) -> Result<Vec<i32>, std::num::ParseIntError> {
    stream::iter(inputs.into_iter().map(|s| s.parse::<i32>())).try_collect().await
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    println!("{:?}", run(vec!["1", "2", "3"]).await);
    println!("{:?}", run(vec!["1", "x", "3"]).await);
}

Ok([1, 2, 3])
Err(ParseIntError { kind: InvalidDigit })

Streams or something simpler?

Use a stream when the items arrive over time and you want to process them one at a time with backpressure: the consumer's pace controls how fast the source is polled. If the data is already in memory, use an iterator. If you only need to receive messages in a loop, while let Some(m) = rx.recv().await is simpler than wrapping the receiver. And if you need to implement a stream over some resource, consider async-stream or the channel-plus-spawned-producer pattern before writing poll_next, which needs pinning and wakers done right; the machinery is explained in futures as state machines, and where the language is heading is in async iterators.