Rust Tokio Streams
A stream is an async iterator whose items arrive over time. Tokio streams carry incoming connections, channel messages, and chunks of a file.
next() gives Some, then None
A Stream is an async iterator: next().await hands you each item as it arrives, and finally None:
let mut stream = tokio_stream::iter(vec![10, 20]);
println!("first = {:?}", stream.next().await);
println!("second = {:?}", stream.next().await);
println!("end = {:?}", stream.next().await);Adapters, and the standard reading loop
The tokio-stream crate provides StreamExt — the adapter methods like map, filter, and take. Adapters are lazy, like iterator adapters: nothing flows until you poll the stream. The standard reading loop is the while let loop:
let mut stream = tokio_stream::iter(1..=10)
.filter(|n| n % 2 == 0)
.map(|n| n * n)
.take(2);
while let Some(n) = stream.next().await {
println!("item: {n}");
}A channel receiver reads like a stream
Channels feel like streams in practice: an mpsc receiver is read item by item, the same way:
let (tx, mut rx) = mpsc::channel::<&str>(2);
tx.send("one message").await.unwrap();
drop(tx);
while let Some(msg) = rx.recv().await {
println!("got: {msg}");
}
println!("channel finished");collect waits for everything
collect is patient. It waits for the last item before your code sees the first one. Every item sits in memory at once.
let all: Vec<i32> = tokio_stream::iter(vec![1, 2, 3]).collect().await;
println!("collected = {all:?}");The loop can walk away early
The while let loop owes the stream nothing. break drops it mid-flow. The source stops being asked for items.
let mut s = tokio_stream::iter(1..=100);
while let Some(n) = s.next().await {
if n == 3 { break; }
println!("used: {n}");
}
println!("stopped early");Example
// Cargo.toml: tokio = { version = "1", features = ["full"] }, tokio-stream = "0.1"
use tokio_stream::StreamExt;
#[tokio::main]
async fn main() {
let mut stream = tokio_stream::iter(vec![1, 2, 3, 4, 5])
.filter(|n| n % 2 == 1)
.map(|n| n * 10);
while let Some(value) = stream.next().await {
println!("item: {}", value);
}
println!("stream finished");
}Key points
- Add it in Cargo.toml: tokio-stream = "0.1" — it is a companion crate, not part of tokio itself
- Process items as they arrive when you can.
- The loop frees each item as it goes.
- collect holds them all at once
- Streams favour handling each item as it arrives.
In this example, tokio_stream::iter turns a plain Vec into a stream — a stand-in for real sources like sockets. filter keeps the odd numbers, and map multiplies them by ten, exactly like the iterator adapters. The while let loop awaits each item as it comes. A real stream would pause between items. This one is instant. But the code shape is identical.
Note
Collect into a Vec only when you truly need everything at once.
Try it Yourself
Edit the code, then press Run.
Running is simulated · Rust Playground
Exercise
What does stream.next().await return when the stream is over?