codeloopback
Guide contents

Lesson 28 of 44

Streams

A stream is the async version of an iterator: a value that produces zero or more items over time, and each item may take an .await to arrive. After this lesson you can create streams from iterators, channels, timers, and generators, transform them with adapters, merge several of them, run async work over their items with bounded concurrency, and reason about cancel safety and backpressure.

Item Crate Role
Stream (poll_next) futures-core, re-exported by tokio-stream and futures the trait
StreamExt tokio-stream or futures (two different traits) adapter methods
ReceiverStream, IntervalStream, BroadcastStream, WatchStream, LinesStream tokio-stream::wrappers turn Tokio types into streams
StreamMap tokio-stream many keyed streams in one
stream::unfold, buffered, buffer_unordered, for_each_concurrent futures building and concurrent processing
stream!, try_stream! async-stream generator syntax

The examples use these crates:

TOMLCargo.toml
[dependencies]
tokio = { version = "1", features = ["full", "test-util"] }
tokio-stream = "0.1"
futures = "0.3"
async-stream = "0.3"

tokio-stream needs its default time feature for timeout, throttle, and chunks_timeout. Examples that involve time use the paused clock (start_paused = true), which makes sleeps finish instantly and in a fixed order. Time: sleep, timeout, and interval explains it.

Example 1: The Stream trait

Stream has one required method, poll_next. It is the Iterator::next of async code. The method returns Poll<Option<Item>>:

Return value Meaning
Poll::Ready(Some(item)) the next item
Poll::Ready(None) the stream is finished
Poll::Pending no item yet; the stream has registered the task's waker and will wake it when an item is ready

Poll, Context, Waker, and Pin come from Polling, Wakers, and a Tiny Executor. This stream never returns Pending, so it never needs the waker. Writing Futures and Streams by Hand covers streams that do.

Rust
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_stream::{Stream, StreamExt};

struct Countdown {
    remaining: u32,
}

impl Stream for Countdown {
    type Item = u32;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.remaining == 0 {
            return Poll::Ready(None);
        }
        let value = self.remaining;
        self.remaining -= 1;
        Poll::Ready(Some(value))
    }
}

#[tokio::main]
async fn main() {
    let mut countdown = Countdown { remaining: 3 };
    while let Some(n) = countdown.next().await {
        println!("{n}");
    }
    println!("done");
}
Output
3
2
1
done

next is the StreamExt method that turns one poll_next call into a future. A stream is not usable with for, because for calls a synchronous next. The loop form is while let Some(item) = stream.next().await.

Example 2: Streams from iterators, and single-item streams

tokio_stream::iter wraps any IntoIterator as a stream that is always ready. once yields one item. empty yields nothing. pending never yields and never ends.

Rust
use std::time::Duration;
use tokio_stream::{self as stream, StreamExt};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut numbers = stream::iter(vec![10, 20, 30]);
    while let Some(n) = numbers.next().await {
        println!("iter: {n}");
    }

    let mut one = stream::once("single");
    println!("once: {:?}", one.next().await);
    println!("once again: {:?}", one.next().await);

    let mut nothing = stream::empty::<u8>();
    println!("empty: {:?}", nothing.next().await);

    let mut never = stream::pending::<u8>();
    let waited = tokio::time::timeout(Duration::from_secs(1), never.next()).await;
    println!("pending timed out: {}", waited.is_err());
}
Output
iter: 10
iter: 20
iter: 30
once: Some("single")
once again: None
empty: None
pending timed out: true

tokio::time::timeout cancels the pending stream's next future after one virtual second. next is cancel safe, so cancelling it loses nothing (Example 23).

Example 3: Adapters that change items

Adapters take a stream and return a new stream. Nothing runs until the final stream is polled. map, filter, filter_map, take, and skip mirror the Iterator methods of the same names.

Rust
use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let mut s = stream::iter(1..=10)
        .skip(1)
        .filter(|n| n % 2 == 0)
        .map(|n| n * n)
        .take(3);

    while let Some(n) = s.next().await {
        println!("{n}");
    }

    let mut parsed = stream::iter(["7", "x", "9"]).filter_map(|s| s.parse::<i32>().ok());
    while let Some(n) = parsed.next().await {
        println!("parsed {n}");
    }
}
Output
4
16
36
parsed 7
parsed 9

The closures are synchronous. To run an async operation for each item, use then (Example 4).

Example 4: then, take_while, and map_while

then takes a closure that returns a future. The stream yields the future's output, and processes one item at a time, in order. take_while ends the stream at the first item that fails a test. map_while maps items and ends the stream when the closure returns None.

Rust
use std::time::Duration;
use tokio::time::sleep;
use tokio_stream::{self as stream, StreamExt};

async fn lookup(id: u32) -> String {
    sleep(Duration::from_millis(10)).await;
    format!("user-{id}")
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut names = stream::iter([1, 2, 3]).then(lookup);
    while let Some(name) = names.next().await {
        println!("{name}");
    }

    let small: Vec<u32> = stream::iter([1, 2, 3, 10, 4])
        .take_while(|n| *n < 5)
        .collect()
        .await;
    println!("take_while: {small:?}");

    let digits: Vec<u32> = stream::iter(["4", "5", "x", "6"])
        .map_while(|s| s.parse().ok())
        .collect()
        .await;
    println!("map_while: {digits:?}");
}
Output

Example 5: Consuming a stream

Consumers return a future instead of a stream: collect, fold, all, any, and next. collect builds a Vec, String, VecDeque, LinkedList, (), or a Result or Option of one of those. Collecting into Result stops at the first Err.

Rust
use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let total = stream::iter(1..=4).fold(0, |acc, n| acc + n).await;
    println!("fold: {total}");

    let mut s = stream::iter([2, 4, 6]);
    println!("all even: {}", s.all(|n| n % 2 == 0).await);
    let mut s = stream::iter([1, 3, 5]);
    println!("any even: {}", s.any(|n| n % 2 == 0).await);

    let text: String = stream::iter(["ab", "cd", "ef"]).collect().await;
    println!("string: {text}");

    let ok: Result<Vec<i32>, String> = stream::iter([Ok(1), Ok(2), Ok(3)]).collect().await;
    println!("ok: {ok:?}");

    let bad: Result<Vec<i32>, String> = stream::iter([Ok(1), Err("bad".to_string()), Ok(3)])
        .collect()
        .await;
    println!("bad: {bad:?}");

    let mut results = stream::iter([Ok(1), Err("boom"), Ok(3)]);
    println!("try_next: {:?}", results.try_next().await);
    println!("try_next: {:?}", results.try_next().await);
}
Output
fold: 10
all even: true
any even: false
string: abcdef
ok: Ok([1, 2, 3])
bad: Err("bad")
try_next: Ok(Some(1))
try_next: Err("boom")

all and any take &mut self, so the stream stays usable afterwards. try_next returns Result<Option<T>, E> for a stream of Result<T, E>, which allows ? in a loop: while let Some(item) = stream.try_next().await? { ... }.

Example 6: Which crate provides which method

Two crates define a trait called StreamExt: tokio_stream::StreamExt and futures::StreamExt. They have overlapping and different methods. The table below was built by reading tokio-stream 0.1.19 (src/stream_ext.rs) and futures-util 0.3.34 (src/stream/stream/mod.rs).

Method tokio-stream futures
next, map, filter, filter_map, then, take, skip, take_while, skip_while, fuse, chain, fold, collect, all, any, peekable yes yes
map_while yes no
merge yes no (futures::stream::select is the equivalent function)
timeout, timeout_repeating yes no
throttle yes no
chunks_timeout yes no
try_next yes (on StreamExt) yes, but on TryStreamExt
buffered, buffer_unordered no yes
for_each, for_each_concurrent no yes
enumerate, zip, scan, inspect, count, cycle, concat, unzip no yes
flatten, flat_map, flatten_unordered no yes
chunks, ready_chunks no yes
take_until, catch_unwind, by_ref, boxed, forward, split no yes

The next example shows what happens when both traits are in scope.

Example 7: Importing both StreamExt traits is ambiguous

When both traits are imported, a call such as .next() matches two methods, and the compiler rejects it.

Rust✗ does not compile · E0034
use futures::StreamExt;
use tokio_stream::StreamExt as _;

#[tokio::main]
async fn main() {
    let mut s = tokio_stream::iter([1, 2, 3]);
    println!("{:?}", s.next().await);
}
Output
error[E0034]: multiple applicable items in scope
 --> src/main.rs:7:24
  |
7 |     println!("{:?}", s.next().await);
  |                        ^^^^ multiple `next` found
  |
  = note: candidate #1 is defined in an impl of the trait `tokio_stream::StreamExt` for the type `St`
  = note: candidate #2 is defined in an impl of the trait `futures::StreamExt` for the type `T`
help: disambiguate the method for candidate #1
  |
7 -     println!("{:?}", s.next().await);
7 +     println!("{:?}", tokio_stream::StreamExt::next(&mut s).await);
  |
help: disambiguate the method for candidate #2
  |
7 -     println!("{:?}", s.next().await);
7 +     println!("{:?}", futures::StreamExt::next(&mut s).await);
  |

The usual arrangement is one StreamExt per file. Import tokio_stream::StreamExt when the file needs timeout, throttle, merge, or chunks_timeout. Import futures::StreamExt when it needs buffer_unordered, for_each_concurrent, enumerate, or zip. A file that needs both can import one under as _ and call the other with a path, such as tokio_stream::StreamExt::timeout(stream, duration) (Example 26).

Example 8: Methods that only futures provides

These calls have no tokio-stream equivalent.

Rust
use futures::{StreamExt, stream};

#[tokio::main]
async fn main() {
    let letters = stream::iter(["a", "b", "c"]);
    let numbered: Vec<(usize, &str)> = letters.enumerate().collect().await;
    println!("{numbered:?}");

    let pairs: Vec<(i32, char)> = stream::iter([1, 2, 3])
        .zip(stream::iter(['x', 'y']))
        .collect()
        .await;
    println!("{pairs:?}");

    let running: Vec<i32> = stream::iter([1, 2, 3, 4])
        .scan(0, |sum, n| {
            *sum += n;
            futures::future::ready(Some(*sum))
        })
        .collect()
        .await;
    println!("{running:?}");

    stream::iter(["x", "y"])
        .for_each(|s| async move { println!("each {s}") })
        .await;
}
Output
[(0, "a"), (1, "b"), (2, "c")]
[(1, 'x'), (2, 'y')]
[1, 3, 6, 10]
each x
each y

zip ends when the shorter stream ends. scan threads mutable state through the items; its closure returns a future.

Example 9: fuse

The Stream contract does not say what happens if poll_next is called again after it returned None. A stream may return Some again, panic, or return None forever. fuse guarantees None forever after the first None. Adapters such as merge and timeout fuse their inputs internally.

Rust
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_stream::{Stream, StreamExt};

struct Flaky {
    step: u32,
}

impl Stream for Flaky {
    type Item = u32;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        self.step += 1;
        match self.step {
            1 => Poll::Ready(Some(1)),
            2 => Poll::Ready(None),
            n => Poll::Ready(Some(n)),
        }
    }
}

#[tokio::main]
async fn main() {
    let mut raw = Flaky { step: 0 };
    println!("raw:   {:?} {:?} {:?}", raw.next().await, raw.next().await, raw.next().await);

    let mut fused = Flaky { step: 0 }.fuse();
    println!("fused: {:?} {:?} {:?}", fused.next().await, fused.next().await, fused.next().await);
}
Output
raw:   Some(1) None Some(3)
fused: Some(1) None None

Example 10: timeout

timeout(duration) wraps each item in Result<T, Elapsed>. The deadline applies to each item. When it passes, the stream yields Err(Elapsed) and continues to poll the inner stream, so a later item still arrives. tokio_stream::Elapsed is the error type.

Rust
use std::time::Duration;
use tokio::time::sleep;
use tokio_stream::StreamExt;

fn delayed() -> impl tokio_stream::Stream<Item = &'static str> {
    async_stream::stream! {
        sleep(Duration::from_millis(50)).await;
        yield "fast";
        sleep(Duration::from_millis(500)).await;
        yield "slow";
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let stream = delayed().timeout(Duration::from_millis(100));
    tokio::pin!(stream);

    while let Some(item) = stream.next().await {
        match item {
            Ok(value) => println!("got {value}"),
            Err(_) => println!("timed out waiting for an item"),
        }
    }
}
Output
got fast
timed out waiting for an item
got slow

To stop at the first timeout instead, add .take_while(Result::is_ok). timeout_repeating(interval) takes an Interval and yields Err(Elapsed) on every tick that passes without an item. The stream is built with async_stream::stream!, explained in Example 19. tokio::pin! is needed because that stream is not Unpin.

Example 11: chunks_timeout

chunks_timeout(max_size, duration) groups items into Vecs. A batch is emitted when it reaches max_size, when duration has passed since the first item of the batch, or when the stream ends. Empty batches are never emitted. It panics if max_size is zero.

Rust
use std::time::Duration;
use tokio::time::sleep;
use tokio_stream::StreamExt;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let source = async_stream::stream! {
        for n in 1..=4 {
            yield n;
        }
        sleep(Duration::from_secs(5)).await;
        yield 5;
        yield 6;
    };

    let batches = source.chunks_timeout(3, Duration::from_secs(2));
    tokio::pin!(batches);

    while let Some(batch) = batches.next().await {
        println!("{batch:?}");
    }
}
Output
[1, 2, 3]
[4]
[5, 6]

Items 1 to 3 fill a batch. Item 4 waits alone until the two-second deadline. Items 5 and 6 arrive after the pause and are flushed when the stream ends. Batching like this reduces the number of database writes or network calls when items arrive faster than the sink can absorb them one by one.

Example 12: throttle

throttle(duration) enforces a minimum gap between items. The timer has a granularity of one millisecond. Under the paused clock, Instant differences are exact, so the elapsed time can be printed.

Rust
use std::time::Duration;
use tokio::time::Instant;
use tokio_stream::{self as stream, StreamExt};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut s = stream::iter(["a", "b", "c"]).throttle(Duration::from_millis(200));

    while let Some(item) = s.next().await {
        println!("{item} at {} ms", start.elapsed().as_millis());
    }
}
Output

Example 13: merge

merge combines two streams into one that yields items as either source produces them. The merged stream ends when both sources have ended. When both are ready, merge alternates which source it polls first, so neither can starve the other.

Rust
use std::time::Duration;
use tokio::time::sleep;
use tokio_stream::{Stream, StreamExt};

fn ticker(name: &'static str, every_ms: u64, count: u32) -> impl Stream<Item = String> {
    async_stream::stream! {
        for i in 1..=count {
            sleep(Duration::from_millis(every_ms)).await;
            yield format!("{name}{i}");
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let merged = ticker("a", 30, 3).merge(ticker("b", 50, 2));
    tokio::pin!(merged);

    while let Some(item) = merged.next().await {
        println!("{item}");
    }
}
Output
a1
b1
a2
a3
b2

The order follows the virtual arrival times: a1 at 30 ms, b1 at 50 ms, a2 at 60 ms, a3 at 90 ms, b2 at 100 ms. To merge more than two streams, chain merge calls, use futures::stream::select_all, or use StreamMap (Example 18) when the consumer must know which source produced each item.

Example 14: ReceiverStream

tokio_stream::wrappers::ReceiverStream turns an mpsc::Receiver into a stream. All adapters then apply to a channel. UnboundedReceiverStream does the same for an unbounded channel. The stream ends when every Sender is dropped and the buffer is empty. mpsc Channels covers the channel.

Rust
use tokio::sync::mpsc;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(4);

    tokio::spawn(async move {
        for n in 1..=5 {
            tx.send(n).await.unwrap();
        }
    });

    let doubled: Vec<i32> = ReceiverStream::new(rx).map(|n| n * 2).collect().await;
    println!("{doubled:?}");
}
Output
[2, 4, 6, 8, 10]

Example 15: IntervalStream and WatchStream

IntervalStream yields an Instant on every tick of a tokio::time::Interval. WatchStream::new yields the current value of a watch channel first and then each change. WatchStream::from_changes skips the current value. A watch channel keeps only the latest value, so a slow consumer of a WatchStream sees fewer items than were sent. oneshot, broadcast, and watch covers the channel.

Rust
use std::time::Duration;
use tokio::sync::watch;
use tokio::time::interval;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::{IntervalStream, WatchStream};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut ticks = IntervalStream::new(interval(Duration::from_secs(1))).take(3);
    let mut count = 0;
    while ticks.next().await.is_some() {
        count += 1;
        println!("tick {count}");
    }

    let (tx, rx) = watch::channel("initial");
    let mut with_current = WatchStream::new(rx.clone());
    let mut changes_only = WatchStream::from_changes(rx);

    println!("with_current: {:?}", with_current.next().await);

    tx.send("v1").unwrap();
    tx.send("v2").unwrap();
    println!("with_current: {:?}", with_current.next().await);
    println!("changes_only: {:?}", changes_only.next().await);
}
Output

Two sends happened before either stream polled again, so both streams saw only v2.

Example 16: BroadcastStream and lag

BroadcastStream yields Result<T, BroadcastStreamRecvError>. When the receiver falls behind by more than the channel capacity, the next item is Err(Lagged(n)), where n is the number of skipped messages. After that, the stream continues with the oldest retained message.

Rust
use tokio::sync::broadcast;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::BroadcastStream;

#[tokio::main]
async fn main() {
    let (tx, rx) = broadcast::channel(2);
    for n in 1..=5 {
        tx.send(n).unwrap();
    }
    drop(tx);

    let mut s = BroadcastStream::new(rx);
    while let Some(item) = s.next().await {
        match item {
            Ok(n) => println!("got {n}"),
            Err(e) => println!("error: {e}"),
        }
    }
}
Output

Example 17: LinesStream

LinesStream wraps the Lines value from BufReader::lines() (see AsyncRead and AsyncWrite) so that adapters apply to lines of text. Its items are io::Result<String>.

Rust
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::LinesStream;

#[tokio::main]
async fn main() {
    let data: &[u8] = b"alpha\nbeta\n\ngamma\n";
    let lines = LinesStream::new(BufReader::new(data).lines());

    let non_empty: Vec<String> = lines
        .filter_map(|line| line.ok())
        .filter(|line| !line.is_empty())
        .map(|line| line.to_uppercase())
        .collect()
        .await;
    println!("{non_empty:?}");
}
Output

Example 18: StreamMap

StreamMap<K, V> holds many streams under keys and yields (K, item) pairs as items become available. Streams can be inserted and removed while it is being polled. A stream that ends is removed automatically. The map ends when it is empty. It scans all entries on every operation, so it suits a small number of streams.

Rust
use std::pin::Pin;
use std::time::Duration;
use tokio::time::sleep;
use tokio_stream::{Stream, StreamExt, StreamMap};

fn numbers(start: u32, gap_ms: u64) -> Pin<Box<dyn Stream<Item = u32>>> {
    Box::pin(async_stream::stream! {
        for n in start..start + 2 {
            sleep(Duration::from_millis(gap_ms)).await;
            yield n;
        }
    })
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut map = StreamMap::new();
    map.insert("fast", numbers(10, 10));
    map.insert("slow", numbers(20, 25));
    println!("streams: {}", map.len());

    while let Some((key, value)) = map.next().await {
        println!("{key} -> {value}");
        if key == "slow" && value == 20 {
            map.remove("fast");
            println!("removed fast, {} left", map.len());
        }
    }
}
Output
streams: 2
fast -> 10
fast -> 11
slow -> 20
removed fast, 1 left
slow -> 21

StreamMap needs Unpin streams, so each one is boxed with Box::pin. The fast stream is removed before it finishes, so 11 never appears.

Example 19: Building streams with unfold and async-stream

futures::stream::unfold(state, f) builds a stream from a state value and an async closure. The closure returns Some((item, next_state)) or None to end. async_stream::stream! is generator syntax: the body is an async block in which yield value produces an item. try_stream! allows ? and yields Result items, ending the stream at the first error.

Rust
use futures::stream;
use std::pin::pin;
use tokio_stream::StreamExt;

fn parse_all(
    inputs: Vec<&'static str>,
) -> impl tokio_stream::Stream<Item = Result<i32, std::num::ParseIntError>> {
    async_stream::try_stream! {
        for text in inputs {
            let n: i32 = text.parse()?;
            yield n;
        }
    }
}

#[tokio::main]
async fn main() {
    let fib = stream::unfold((0u64, 1u64), |(a, b)| async move { Some((a, (b, a + b))) });
    let first: Vec<u64> = fib.take(8).collect().await;
    println!("unfold: {first:?}");

    let evens = async_stream::stream! {
        for n in 0..10 {
            if n % 2 == 0 {
                yield n;
            }
        }
    };
    let mut evens = pin!(evens);
    while let Some(n) = evens.next().await {
        println!("stream!: {n}");
    }

    let mut results = pin!(parse_all(vec!["1", "2", "x", "4"]));
    while let Some(item) = results.next().await {
        match item {
            Ok(n) => println!("ok {n}"),
            Err(e) => println!("err {e}"),
        }
    }
}
Output
unfold: [0, 1, 1, 2, 3, 5, 8, 13]
stream!: 0
stream!: 2
stream!: 4
stream!: 6
stream!: 8
ok 1
ok 2
err invalid digit found in string

The stream! result is not Unpin, so the example pins it with std::pin::pin! before calling next. Box::pin(stream) also works and produces an Unpin value. take in the first part consumes fib by value, so no pinning is needed there: collect takes ownership and pins internally.

Example 20: Returning streams from functions

A function can return impl Stream<Item = T>. To pass the stream to tokio::spawn, the type must also be Send and 'static. Add + Send to the return type to state that promise. Pin<Box<dyn Stream<Item = T> + Send>> is the form for a value that could be one of several stream types.

Rust
use std::pin::Pin;
use tokio_stream::{self as stream, Stream, StreamExt};

fn evens() -> impl Stream<Item = u32> + Send {
    stream::iter(0..).filter(|n| n % 2 == 0)
}

fn choose(reverse: bool) -> Pin<Box<dyn Stream<Item = u32> + Send>> {
    if reverse {
        Box::pin(stream::iter([3, 2, 1]))
    } else {
        Box::pin(stream::iter([1, 2, 3]))
    }
}

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        let mut s = evens().take(4);
        let mut out = Vec::new();
        while let Some(n) = s.next().await {
            out.push(n);
        }
        out
    });
    println!("{:?}", handle.await.unwrap());

    let mut s = choose(true);
    while let Some(n) = s.next().await {
        println!("boxed {n}");
    }
}
Output
[0, 2, 4, 6]
boxed 3
boxed 2
boxed 1

The boxed stream is Unpin (a Pin<Box<T>> always is), so next works without further pinning.

Example 21: buffered and buffer_unordered

A stream of futures can run those futures concurrently. futures::StreamExt::buffered(n) polls up to n futures at once and yields results in the original order. buffer_unordered(n) yields each result as it completes. Both pull from the source only when they have room, so n is a hard limit on concurrent work.

Rust
use futures::StreamExt;
use std::time::Duration;
use tokio::time::sleep;

async fn work(id: u64) -> u64 {
    sleep(Duration::from_millis(100 - id * 30)).await;
    println!("  finished {id}");
    id
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    println!("buffered(3):");
    let ordered: Vec<u64> = futures::stream::iter(0..3).map(work).buffered(3).collect().await;
    println!("  result order: {ordered:?}");

    println!("buffer_unordered(3):");
    let unordered: Vec<u64> = futures::stream::iter(0..3)
        .map(work)
        .buffer_unordered(3)
        .collect()
        .await;
    println!("  result order: {unordered:?}");
}
Output
buffered(3):
  finished 2
  finished 1
  finished 0
  result order: [0, 1, 2]
buffer_unordered(3):
  finished 2
  finished 1
  finished 0
  result order: [2, 1, 0]

Task 2 sleeps the shortest time and finishes first in both runs. buffered holds its result until tasks 0 and 1 are also done, then yields 0, 1, 2. buffer_unordered yields 2, 1, 0. The map call only creates the futures; the buffer adapter polls them.

Example 22: Bounded concurrency with for_each_concurrent

for_each_concurrent(limit, f) runs an async closure for each item with at most limit running at once. It returns a future that completes when the stream is exhausted and all closures finished. A limit of None means no limit. The closure's futures all run inside the calling task, not on separate tasks, so they share one thread. JoinSet and Structured Concurrency covers the alternative that spawns tasks.

Rust
use futures::StreamExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let running = AtomicUsize::new(0);
    let peak = AtomicUsize::new(0);

    futures::stream::iter(0..10)
        .for_each_concurrent(3, |_id| async {
            let now = running.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            sleep(Duration::from_millis(10)).await;
            running.fetch_sub(1, Ordering::SeqCst);
        })
        .await;

    println!("peak concurrency: {}", peak.load(Ordering::SeqCst));
}
Output
peak concurrency: 3

Example 23: Cancel safety of next

StreamExt::next is cancel safe. Its future holds only a reference to the stream. Dropping the future before it completes does not consume an item, so the next call still receives it. This makes next safe as a branch of select! inside a loop. select!: Racing Futures covers the macro.

Rust
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for n in 1..=3 {
            sleep(Duration::from_millis(250)).await;
            tx.send(n).await.unwrap();
        }
    });

    let mut stream = ReceiverStream::new(rx);
    let mut cancelled = 0;
    loop {
        tokio::select! {
            item = stream.next() => match item {
                Some(n) => println!("item {n}"),
                None => break,
            },
            _ = sleep(Duration::from_millis(100)) => cancelled += 1,
        }
    }
    println!("next() futures dropped before completing: {cancelled}");
}
Output
item 1
item 2
item 3
next() futures dropped before completing: 6

The 100 ms branch wins several times per item. Each win drops the pending next() future. All three items still arrive, in order.

Example 24: Streams are lazy and pull-based

An adapter polls its source only when its own consumer asks for an item. A stream that nobody polls does no work. The counter below shows that take(2) on a chain of adapters pulls only as many source items as needed and stops.

Rust
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio_stream::{self as stream, StreamExt};

static PULLED: AtomicUsize = AtomicUsize::new(0);

#[tokio::main]
async fn main() {
    let source = stream::iter(1..=1_000_000).map(|n| {
        PULLED.fetch_add(1, Ordering::SeqCst);
        n
    });
    let pipeline = source.filter(|n| n % 3 == 0).take(2);
    println!("pulled before polling: {}", PULLED.load(Ordering::SeqCst));

    let items: Vec<i32> = pipeline.collect().await;
    println!("items: {items:?}");
    println!("pulled after: {}", PULLED.load(Ordering::SeqCst));
}
Output
pulled before polling: 0
items: [3, 6]
pulled after: 6

Six items were pulled to find the first two multiples of 3. This pull model is the base of backpressure: a slow consumer polls less often, so the producer is asked for less.

Example 25: Backpressure with a bounded channel

A pull-based stream slows its producer only if the producer is connected through something that can say "full". A bounded mpsc channel does that: send(..).await waits when the buffer is full, and try_send returns an error. An unbounded channel never says "full", so a producer faster than its consumer makes memory grow without limit.

Rust
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TrySendError;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(2);

    let mut accepted = 0;
    let mut rejected = 0;
    for n in 1..=5 {
        match tx.try_send(n) {
            Ok(()) => accepted += 1,
            Err(TrySendError::Full(_)) => rejected += 1,
            Err(TrySendError::Closed(_)) => unreachable!(),
        }
    }
    println!("accepted {accepted}, rejected {rejected}");

    let mut stream = ReceiverStream::new(rx);
    println!("first: {:?}", stream.next().await);
    println!("try_send after one read: {:?}", tx.try_send(6).is_ok());
}
Output
accepted 2, rejected 3
first: Some(1)
try_send after one read: true

Capacity 2 accepts two items and rejects the other three. Reading one item frees one slot, so the next try_send succeeds. A producer that uses send(..).await would wait instead of failing. Backpressure and Load Shedding builds on this.

Example 26: A pipeline with bounded concurrency and batching

This example combines the tools above: a stream of ids, concurrent lookups with buffer_unordered, and batching with chunks_timeout. Both extension traits are needed, so one is imported anonymously and the other is called by path.

Rust
use futures::StreamExt as _;
use std::time::Duration;
use tokio::time::sleep;

async fn fetch(id: u32) -> (u32, u32) {
    sleep(Duration::from_millis(u64::from(50 - id * 10))).await;
    (id, id * id)
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let results = futures::stream::iter(1..=4).map(fetch).buffer_unordered(2);

    let batches = tokio_stream::StreamExt::chunks_timeout(results, 3, Duration::from_secs(1));
    tokio::pin!(batches);

    while let Some(batch) = batches.next().await {
        println!("batch: {batch:?}");
    }
}
Output
batch: [(2, 4), (1, 1), (3, 9)]
batch: [(4, 16)]

use futures::StreamExt as _ brings the trait's methods into scope without a name, so map, buffer_unordered, and next resolve to futures. The tokio-stream method is called by its full path, so there is no ambiguity.

Example 27: Sink, named only

Sink is the write-side counterpart of Stream: a value that accepts items with start_send and pushes them out with poll_flush. It lives in futures-sink, and futures::SinkExt adds send, feed, flush, and close. Framed and FramedWrite in Codecs with tokio-util implement Sink, and that lesson uses SinkExt::send. Nothing else in this lesson needs it.

Interview points

  • A Stream is an async iterator: poll_next returns Ready(Some), Ready(None), or Pending. for loops do not work with it. The loop is while let Some(x) = s.next().await.
  • tokio_stream::StreamExt and futures::StreamExt are different traits. Importing both makes overlapping methods ambiguous. timeout, throttle, merge, chunks_timeout, and map_while exist only in tokio-stream. buffered, buffer_unordered, for_each_concurrent, enumerate, and zip exist only in futures.
  • next requires Unpin. Use tokio::pin!, std::pin::pin!, or Box::pin for streams from stream! or unfold.
  • next is cancel safe. It can be a select! branch in a loop without losing items.
  • buffered(n) preserves input order. buffer_unordered(n) yields completion order. Both limit concurrency to n.
  • for_each_concurrent runs its futures inside one task. Use JoinSet or spawn for parallelism across threads.
  • StreamMap yields keyed items and drops finished streams silently. Its cost is linear in the number of streams.
  • BroadcastStream reports lag as Err(Lagged(n)). WatchStream coalesces values.
  • Bounded channels give backpressure. Unbounded channels grow until memory runs out.
  • fuse guarantees that a finished stream stays finished.

Summary

  • Stream::poll_next returns Poll<Option<Item>>; None ends the stream.
  • stream::iter, once, empty, and pending build simple streams. unfold and async_stream::stream! build stateful ones.
  • Adapters are lazy. Nothing runs until the final stream is polled, and each adapter pulls from its source only on demand.
  • collect builds Vec, String, VecDeque, LinkedList, (), and Result or Option of those.
  • timeout wraps items in Result<T, Elapsed> and continues after a timeout. chunks_timeout batches by size or time. throttle spaces items.
  • merge combines two streams fairly. StreamMap merges keyed streams that can be added and removed while polling.
  • Wrappers ReceiverStream, IntervalStream, BroadcastStream, WatchStream, and LinesStream connect Tokio types to the adapter methods.
  • buffer_unordered and for_each_concurrent bound concurrent work. Bounded channels bound queued work.
  • Sink is the write-side trait, used with SinkExt in the codecs lesson.