Stream in Rust: Sources & vs Iterator Explained

TL;DR: A Stream is the async equivalent of Iterator; it produces a sequence of values over time, yielding each one asynchronously. Where an iterator returns Option<Item> synchronously, a stream returns Poll<Option<Item>>; it may not have the next value yet and will wake up the executor when it does. Streams model things like database result sets, WebSocket messages, file line-by-line reads, and SSE event feeds.

What Is a Stream?

Stream is a trait that produces zero or more values asynchronously; like an async Iterator. Each call to poll_next returns Poll::Ready(Some(item)), Poll::Ready(None) (end of stream), or Poll::Pending (not ready yet).

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

You rarely implement Stream directly. Instead, you use combinators from tokio_stream or futures::StreamExt the same way you use Iterator adapters.

How Do You Consume a Stream?

Use the StreamExt trait from tokio_stream or futures; it adds .next(), .map(), .filter(), .collect() and more to any Stream.

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"

use tokio_stream::{self as stream, StreamExt};
 
#[tokio::main]
async fn main() {
    // Create a stream from an iterator
    let mut s = stream::iter(vec![1, 2, 3, 4, 5]);
 
    // Consume with while let; like a for loop for async
    while let Some(val) = s.next().await {
        println!("{val}");
    }
 
    // Or use combinators
    let sum: i32 = stream::iter(0..10)
        .filter(|x| std::future::ready(x % 2 == 0))
        .map(|x| x * x)
        .collect::<Vec<_>>()
        .await
        .iter()
        .sum();
 
    println!("sum of even squares: {sum}"); // 0+4+16+36+64 = 120
}

How Do You Create a Stream?

Use async_stream::stream! macro for generator-style streams, or tokio_stream::wrappers for wrapping existing async primitives.

[dependencies]
async-stream = "0.3"
tokio-stream = "0.1"

use async_stream::stream;
use tokio_stream::StreamExt;
 
fn fibonacci() -> impl tokio_stream::Stream<Item = u64> {
    stream! {
        let (mut a, mut b) = (0u64, 1u64);
        loop {
            yield a;
            (a, b) = (b, a + b);
        }
    }
}
 
#[tokio::main]
async fn main() {
    let mut fib = std::pin::pin!(fibonacci());
    for _ in 0..10 {
        if let Some(n) = fib.next().await {
            print!("{n} "); // 0 1 1 2 3 5 8 13 21 34
        }
    }
}

What Are Common Stream Sources?

Tokio and ecosystem crates expose many stream sources; channels, timers, file reads, database cursors, and more.

use tokio_stream::StreamExt;
use tokio::sync::mpsc;
 
// mpsc receiver as a stream
let (tx, rx) = mpsc::channel::<String>(32);
let mut stream = tokio_stream::wrappers::ReceiverStream::new(rx);
 
tokio::spawn(async move {
    tx.send("hello".to_string()).await.unwrap();
    tx.send("world".to_string()).await.unwrap();
});
 
while let Some(msg) = stream.next().await {
    println!("{msg}");
}

Other common stream sources:

• sqlx::query_as!(...).fetch(&pool); database rows as a stream
• tokio::fs::read_dir; directory entries
• reqwest::Response::bytes_stream(); chunked HTTP response body
• tokio_stream::wrappers::IntervalStream; timer ticks

How Is Stream Different From Iterator?

Iterator is synchronous; .next() blocks until a value is available. Stream is async; .next().await suspends the task until the value is ready, freeing the executor to run other tasks.

Frequently Asked Questions

Sources

• tokio-stream crate: Tokio's stream utilities
• futures::StreamExt: Full adapter set
• async-stream crate: Generator-style streams

Related Glossary Terms

• Future: A Future produces one value; a Stream produces many values over time
• Iterator: Stream is the async equivalent of Iterator
• Channel: ReceiverStream bridges mpsc channels into a Stream
• Async/Await: Streams are consumed with .next().await in async contexts
• Tokio: Provides tokio_stream and async stream sources
• Select: select! is commonly used to poll multiple streams or stream-like receivers concurrently
• MPSC: tokio::sync::mpsc receivers are frequently adapted into streams in async systems
• Pin: Many stream combinators require pinned streams before they can be polled safely

Keep Reading

• Building AI Agents in Rust: streaming LLM responses with async Streams
• Rust on AWS Lambda: streaming responses from serverless functions
• Rust Ollama Local LLM Integration: streaming tokens from a local LLM in Rust