Streams

Stream is the async cousin of Iterator: it produces a sequence of items over time, driven by .await instead of .next(). Use streams for paginated APIs, WebSockets, file chunks, and event feeds.

Recipe

Quick-reference recipe card - copy-paste ready.

use tokio_stream::{self as stream, StreamExt};
 
#[tokio::main]
async fn main() {
    let mut s = stream::iter(vec![10, 20, 30]);
    while let Some(n) = s.next().await {
        println!("{n}");
    }
}

When to reach for this: Consuming paginated HTTP, database cursors, Kafka/NATS messages, or SSE/WebSocket events.

Working Example

use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
async fn produce(tx: mpsc::Sender<i32>) {
    for i in 1..=5 {
        tx.send(i).await.unwrap();
    }
}
 
#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(8);
    tokio::spawn(produce(tx));
 
    let mut stream = ReceiverStream::new(rx);
    let sum: i32 = stream.map(|n| n * 2).sum().await;
    println!("{sum}");
}

What this demonstrates:

• Channel receiver adapts to Stream via ReceiverStream.
• StreamExt provides map, filter, take, buffer_unordered.
• sum().await consumes the stream.

Deep Dive

How It Works

• Stream::poll_next: Like Future::poll but yields Option<Item> until None.
• Backpressure: Slow consumers naturally apply pressure on bounded channels.
• Fusion with futures: Many apps map streams to responses (Axum SSE, gRPC streaming).
• StreamExt: Ergonomic methods from futures / tokio-stream crates.

Common Adapters

Rust Notes

use futures::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
 
struct Counter {
    n: u32,
    max: u32,
}
 
impl Stream for Counter {
    type Item = u32;
 
    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.n >= self.max {
            Poll::Ready(None)
        } else {
            let v = self.n;
            self.n += 1;
            Poll::Ready(Some(v))
        }
    }
}

• Manual Stream impls mirror Future pinning rules.
• Prefer adapters over manual impls in application code.
• try_stream! macro builds fallible streams ergonomically.

Gotchas

• Unbounded production - Fast producer overwhelms memory without bounds. Fix: Bounded mpsc or stream::iter with rate limits.
• Forgetting to drive stream - Creating stream without consuming leaks tasks. Fix: Always drain, cancel, or drop explicitly.
• Blocking in stream adapters - map closures must not block. Fix: then with async offload.
• Mixing Stream crates - Import StreamExt from the crate matching your stream type. Fix: use tokio_stream::StreamExt vs futures::StreamExt (enable features).
• No for await in stable Rust - Use while let Some or StreamExt::next. Fix: Loop with .await on next().

Alternatives

FAQs

Related

• Async Basics - async iteration intro
• Combining Futures - select with streams
• Channels in Tokio - ReceiverStream
• Async I/O - byte streams
• Cancellation & Timeouts - stream timeouts

Stack versions: This page was written for Rust 1.97.0 (edition 2024), Tokio 1.x, Axum 0.8, serde 1.0, sqlx 0.8, clap 4, and Polars 0.46+.