Async Streams

An Iterator produces values one by one, and each next() returns immediately. A Stream is the async version: each next() is a future, because the next value may not exist yet. A page of results from an API, a message from a socket, a tick from a timer: all naturally streams.

Concept	Rust
synchronous sequence	`Iterator<Item = T>`
asynchronous sequence	`Stream<Item = T>`
map, filter, take	`stream.map(...)`, `filter`, `take` (same names as `Iterator`)
concurrent mapping	`stream.map(f).buffer_unordered(n)`
sequential async mapping	`stream.then(f)`
batching	`stream.chunks(n)`
merging	`futures::stream::select(a, b)`
backpressure	built in: a stream only produces when polled

Stream isn't in the standard library yet. The ecosystem uses the Stream trait from the futures crate, with extension methods from futures::StreamExt (and a few extras in tokio_stream::StreamExt). All examples use futures, tokio, and tokio-stream.

1. Consuming a stream

stream::iter turns any iterator into a stream. next().await returns Some(item) or None when the stream ends, so the basic loop is while let Some(x) = stream.next().await.

Rust

use futures::StreamExt;
use futures::stream;

#[tokio::main]
async fn main() {
    let mut orders = stream::iter(vec!["order-1", "order-2", "order-3"]);

    while let Some(order) = orders.next().await {
        println!("processing {order}");
    }
}

Output

processing order-1
processing order-2
processing order-3

There's no for loop over streams yet (an async for syntax is being designed), so while let is the idiom.

2. Adapters look just like `Iterator`

map, filter, take, skip, enumerate, and collect behave like their iterator namesakes. Note that filter's predicate returns a future (it's allowed to await), so a sync condition is wrapped in async { ... } or std::future::ready(...).

Rust

use futures::StreamExt;
use futures::stream;
use std::future::ready;

#[tokio::main]
async fn main() {
    let prices: Vec<String> = stream::iter(1..=10)
        .filter(|n| ready(n % 2 == 0))
        .map(|n| n * 150)
        .take(3)
        .map(|cents| format!("${}.{:02}", cents / 100, cents % 100))
        .collect()
        .await;

    println!("{prices:?}");
}

Output

["$3.00", "$6.00", "$9.00"]

3. `then`: an async map

map takes a normal function. When each item needs async work (a lookup, a network call), use then, whose closure returns a future. Items are processed one at a time, in order.

Rust

use futures::StreamExt;
use futures::stream;
use std::time::Duration;

async fn lookup_name(user_id: u32) -> String {
    tokio::time::sleep(Duration::from_millis(10)).await;
    format!("user-{user_id}")
}

#[tokio::main]
async fn main() {
    let names: Vec<String> = stream::iter([3, 1, 2]).then(lookup_name).collect().await;
    println!("{names:?}");
}

Output

["user-3", "user-1", "user-2"]

4. `filter_map` with async work

filter_map's closure returns a future of Option: Some(x) keeps (a transformed) item, None drops it. It's filtering and mapping in one step, with async allowed.

Rust

use futures::StreamExt;
use futures::stream;

async fn fetch_email(user_id: u32) -> Option<String> {
    tokio::task::yield_now().await; // database lookup
    match user_id {
        1 => Some(String::from("ana@example.com")),
        3 => Some(String::from("cy@example.com")),
        _ => None, // no email on file
    }
}

#[tokio::main]
async fn main() {
    let emails: Vec<String> = stream::iter(1..=4).filter_map(fetch_email).collect().await;
    println!("{emails:?}");
}

Output

["ana@example.com", "cy@example.com"]

5. `fold` and `for_each`

fold reduces a stream to one value with an async step; for_each runs an async action for every item. Both return a future that completes when the stream ends.

Rust

use futures::StreamExt;
use futures::stream;

#[derive(Debug, Default)]
struct Totals {
    count: u32,
    cents: u64,
}

#[tokio::main]
async fn main() {
    let payments = stream::iter([("ana", 2500u64), ("ben", 1200), ("ana", 800)]);

    let totals = payments
        .clone()
        .fold(Totals::default(), |mut acc, (_, cents)| async move {
            acc.count += 1;
            acc.cents += cents;
            acc
        })
        .await;
    println!("{totals:?}");

    payments
        .for_each(|(who, cents)| async move {
            println!("receipt sent to {who} for {cents}");
        })
        .await;
}

Output

Totals { count: 3, cents: 4500 }
receipt sent to ana for 2500
receipt sent to ben for 1200
receipt sent to ana for 800

6. Streams are lazy and pull-based

Like iterators, streams do nothing until something polls them, and a stream produces its next item only when the consumer asks for it. That's backpressure for free: a slow consumer automatically slows the producer. stream::unfold builds a stream from a state and an async step function, and the log shows the interleaving.

Rust

use futures::StreamExt;
use futures::stream;

#[tokio::main]
async fn main() {
    let numbers = stream::unfold(1, |n| async move {
        if n > 3 {
            return None; // end of stream
        }
        println!("  producing {n}");
        Some((n, n + 1)) // (item to yield, next state)
    });

    println!("stream created, nothing produced yet");
    let mut numbers = Box::pin(numbers);
    while let Some(n) = numbers.next().await {
        println!("consumed {n}");
    }
}

Output

stream created, nothing produced yet
  producing 1
consumed 1
  producing 2
consumed 2
  producing 3
consumed 3

(The Box::pin is explained in example 22.)

7. `unfold` for pagination

A paginated API is the classic stream: fetch a page, yield its items, remember the next cursor, stop when there's no next page. unfold holds the cursor as its state, and flat_map flattens the pages into individual items.

Rust

use futures::StreamExt;
use futures::stream;

struct Page {
    items: Vec<String>,
    next: Option<u32>,
}

async fn fetch_page(cursor: u32) -> Page {
    tokio::task::yield_now().await; // HTTP GET /users?cursor=...
    let items = (cursor..cursor + 3).map(|i| format!("user{i}")).collect();
    Page { items, next: if cursor < 6 { Some(cursor + 3) } else { None } }
}

fn all_users() -> impl futures::Stream<Item = String> {
    stream::unfold(Some(0), |cursor| async move {
        let page = fetch_page(cursor?).await; // `?` on None ends the stream
        Some((page.items, page.next))
    })
    .flat_map(stream::iter)
}

#[tokio::main]
async fn main() {
    let users: Vec<String> = all_users().collect().await;
    println!("{} users: {users:?}", users.len());
}

Output

9 users: ["user0", "user1", "user2", "user3", "user4", "user5", "user6", "user7", "user8"]

The caller sees a flat stream of users and never deals with cursors.

8. Stopping early saves work

Because streams are lazy, take(n) or take_while stops pulling as soon as it has enough, and later pages are never fetched.

Rust

use futures::StreamExt;
use futures::stream;
use std::sync::atomic::{AtomicU32, Ordering};

static PAGES_FETCHED: AtomicU32 = AtomicU32::new(0);

async fn fetch_page(n: u32) -> Vec<u32> {
    PAGES_FETCHED.fetch_add(1, Ordering::Relaxed);
    tokio::task::yield_now().await;
    (n * 10..n * 10 + 10).collect()
}

#[tokio::main]
async fn main() {
    let first_12: Vec<u32> = stream::iter(0..1000) // up to 1000 pages available
        .then(fetch_page)
        .flat_map(stream::iter)
        .take(12)
        .collect()
        .await;

    println!("{first_12:?}");
    println!("pages fetched: {}", PAGES_FETCHED.load(Ordering::Relaxed));
}

Output

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
pages fetched: 2

9. Other ways to create streams

futures::stream has constructors for the common cases: once (one item from a future), repeat_with (endless, from a closure), empty, and iter. chain concatenates streams.

Rust

use futures::StreamExt;
use futures::stream;

#[tokio::main]
async fn main() {
    let header = stream::once(async { String::from("id,name") });
    let mut next_id = 0;
    let rows = stream::repeat_with(move || {
        next_id += 1;
        format!("{next_id},user{next_id}")
    })
    .take(3);
    let footer = stream::empty::<String>();

    let csv: Vec<String> = header.chain(rows).chain(footer).collect().await;
    println!("{}", csv.join("\n"));
}

Output

id,name
1,user1
2,user2
3,user3

10. A channel as a stream

tokio_stream::wrappers::ReceiverStream turns an mpsc::Receiver into a stream, so everything produced by other tasks can flow through stream adapters.

Rust

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

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

    tokio::spawn(async move {
        for temp in [21.5, 22.0, 35.2, 21.8, 40.1] {
            tx.send(temp).await.unwrap();
        }
    });

    let alerts: Vec<String> = ReceiverStream::new(rx)
        .enumerate()
        .filter(|(_, t)| std::future::ready(*t > 30.0))
        .map(|(i, t)| format!("reading #{i}: {t} C is too hot"))
        .collect()
        .await;
    println!("{alerts:#?}");
}

Output

[
    "reading #2: 35.2 C is too hot",
    "reading #4: 40.1 C is too hot",
]

11. A timer as a stream

IntervalStream wraps tokio::time::interval. Combined with take and zip, periodic work becomes a stream pipeline.

Rust

use futures::StreamExt;
use futures::stream;
use std::time::Duration;
use tokio_stream::wrappers::IntervalStream;

#[tokio::main]
async fn main() {
    let ticks = IntervalStream::new(tokio::time::interval(Duration::from_millis(20)));
    let jobs = stream::iter(["backup", "vacuum", "reindex"]);

    // One job per tick: zip pairs them up and ends with the shorter stream.
    let log: Vec<String> = jobs.zip(ticks).enumerate().map(|(i, (job, _tick))| format!("tick {i}: run {job}")).collect().await;
    println!("{log:#?}");
}

Output

[
    "tick 0: run backup",
    "tick 1: run vacuum",
    "tick 2: run reindex",
]

12. `buffered`: concurrent, in order

then processes one item at a time. To have several futures in flight, map each item to a future and apply buffered(n): up to n run concurrently, and results come out in the original order.

Rust

use futures::StreamExt;
use futures::stream;
use std::time::{Duration, Instant};

async fn fetch(id: u32) -> String {
    tokio::time::sleep(Duration::from_millis(100 - id as u64 * 20)).await; // later ids are faster
    format!("item-{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let items: Vec<String> = stream::iter(0..4).map(fetch).buffered(4).collect().await;
    println!("{items:?}");
    println!("concurrent (under 200 ms): {}", start.elapsed() < Duration::from_millis(200));
}

Output

["item-0", "item-1", "item-2", "item-3"]
concurrent (under 200 ms): true

13. `buffer_unordered`: concurrent, as they finish

buffer_unordered(n) also runs up to n futures at once, but yields each result as soon as it's ready. It's the most common way to limit concurrency over a list of work.

Rust

use futures::StreamExt;
use futures::stream;
use std::time::Duration;

async fn fetch(id: u32) -> String {
    tokio::time::sleep(Duration::from_millis(100 - id as u64 * 30)).await; // later ids are faster
    format!("item-{id}")
}

#[tokio::main]
async fn main() {
    let items: Vec<String> = stream::iter(0..4).map(fetch).buffer_unordered(4).collect().await;
    println!("{items:?}");
}

Output

["item-3", "item-2", "item-1", "item-0"]

14. Limiting concurrency with `buffer_unordered`

With a limit smaller than the number of items, at most n futures run at the same time. This example tracks the peak to show it.

Rust

use futures::StreamExt;
use futures::stream;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);
static PEAK: AtomicUsize = AtomicUsize::new(0);

async fn call_api(id: u32) -> u32 {
    let now = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1;
    PEAK.fetch_max(now, Ordering::SeqCst);
    tokio::time::sleep(Duration::from_millis(10)).await;
    IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
    id * 2
}

#[tokio::main]
async fn main() {
    let results: Vec<u32> = stream::iter(1..=20).map(call_api).buffer_unordered(4).collect().await;
    println!("{} results, sum {}", results.len(), results.iter().sum::<u32>());
    println!("peak concurrency: {}", PEAK.load(Ordering::SeqCst));
}

Output

20 results, sum 420
peak concurrency: 4

This is usually simpler than a Semaphore when the work is a list of items.

15. `for_each_concurrent`

When you don't need results back, for_each_concurrent(limit, f) runs f for every item with at most limit in flight.

Rust

use futures::StreamExt;
use futures::stream;
use std::sync::Mutex;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let sent = Mutex::new(Vec::new());

    stream::iter(["ana", "ben", "cy", "dee", "eve"])
        .for_each_concurrent(2, |user| {
            let sent = &sent;
            async move {
                tokio::time::sleep(Duration::from_millis(10)).await; // send a newsletter
                sent.lock().unwrap().push(user);
            }
        })
        .await;

    let mut sent = sent.into_inner().unwrap();
    sent.sort();
    println!("newsletter sent to {sent:?}");
}

Output

newsletter sent to ["ana", "ben", "cy", "dee", "eve"]

16. Batching with `chunks`

chunks(n) groups items into Vecs of up to n (the last one may be shorter). It's how you turn a stream of rows into batch inserts.

Rust

use futures::StreamExt;
use futures::stream;

async fn insert_batch(rows: Vec<u32>) -> usize {
    tokio::task::yield_now().await; // INSERT ... VALUES (...), (...), ...
    println!("inserted batch {rows:?}");
    rows.len()
}

#[tokio::main]
async fn main() {
    let inserted: usize = stream::iter(1..=8).chunks(3).then(insert_batch).fold(0, |a, n| async move { a + n }).await;
    println!("total rows: {inserted}");
}

Output

inserted batch [1, 2, 3]
inserted batch [4, 5, 6]
inserted batch [7, 8]
total rows: 8

17. `chunks_timeout`: batch by size or time

For live streams, waiting to fill a batch could take forever. tokio_stream::StreamExt::chunks_timeout(n, duration) emits a batch when it's full or when the time runs out, whichever comes first.

Rust

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

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(16);
    tokio::spawn(async move {
        for i in 1..=4 {
            tx.send(i).await.unwrap(); // a burst of 4...
        }
        tokio::time::sleep(Duration::from_millis(150)).await; // ...then a pause...
        tx.send(5).await.unwrap(); // ...then one more
    });

    let batches = ReceiverStream::new(rx).chunks_timeout(3, Duration::from_millis(50));
    tokio::pin!(batches);
    while let Some(batch) = batches.next().await {
        println!("flush {batch:?}");
    }
}

Output

flush [1, 2, 3]
flush [4]
flush [5]

futures::StreamExt and tokio_stream::StreamExt both define next, map, and friends, so import only the one you need in a given module to avoid ambiguity.

18. Merging streams

futures::stream::select(a, b) merges two streams, yielding items from whichever has one ready. select_all does the same for a list of streams.

Rust

use futures::StreamExt;
use futures::stream::{self, select, select_all};
use std::time::Duration;

fn sensor(name: &'static str, every_ms: u64, count: usize) -> impl futures::Stream<Item = String> {
    stream::iter(0..count).then(move |i| async move {
        tokio::time::sleep(Duration::from_millis(every_ms)).await;
        format!("{name}:{i}")
    })
}

#[tokio::main]
async fn main() {
    let mut merged: Vec<String> = select(Box::pin(sensor("fast", 10, 3)), Box::pin(sensor("slow", 25, 2))).collect().await;
    merged.sort();
    println!("{merged:?}");

    let all = select_all(["a", "b", "c"].map(|n| Box::pin(sensor(n, 5, 2))));
    println!("{} readings from 3 sensors", all.count().await);
}

Output

["fast:0", "fast:1", "fast:2", "slow:0", "slow:1"]
6 readings from 3 sensors

19. `take_while`, `skip_while`, and `take_until`

take_while ends the stream at the first item that fails a condition; skip_while drops items until the condition fails. take_until(future) ends the stream when some other future completes, like a shutdown signal.

Rust

use futures::StreamExt;
use futures::stream;
use std::future::ready;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let log = ["DEBUG boot", "DEBUG config", "INFO started", "INFO request", "ERROR crash", "INFO restart"];

    let after_boot: Vec<&str> = stream::iter(log)
        .skip_while(|l| ready(l.starts_with("DEBUG")))
        .take_while(|l| ready(!l.starts_with("ERROR")))
        .collect()
        .await;
    println!("{after_boot:?}");

    // An endless ticker, stopped after ~55 ms by a timer.
    let ticks = stream::unfold(0u32, |n| async move {
        tokio::time::sleep(Duration::from_millis(20)).await;
        Some((n, n + 1))
    });
    let seen: Vec<u32> = ticks.take_until(tokio::time::sleep(Duration::from_millis(55))).collect().await;
    println!("ticks before shutdown: {seen:?}");
}

Output

["INFO started", "INFO request"]
ticks before shutdown: [0, 1]

20. Timeouts and throttling

tokio_stream::StreamExt adds time-based adapters: timeout(d) yields Err(Elapsed) for an item that took too long to arrive (and keeps going), and throttle(d) spaces items at least d apart.

Rust

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

#[tokio::main]
async fn main() {
    let slow_source = stream::iter([10u64, 80, 10]).then(|ms| async move {
        tokio::time::sleep(Duration::from_millis(ms)).await;
        ms
    });
    let results: Vec<_> = slow_source.timeout(Duration::from_millis(50)).map(|r| r.map_err(|_| "timeout")).collect().await;
    println!("{results:?}");

    let start = Instant::now();
    let throttled: Vec<u32> = stream::iter(1..=4).throttle(Duration::from_millis(30)).collect().await;
    println!("{throttled:?} took at least 90 ms: {}", start.elapsed() >= Duration::from_millis(90));
}

Output

[Ok(10), Err("timeout"), Ok(80), Ok(10)]
[1, 2, 3, 4] took at least 90 ms: true

The timed-out item isn't lost: timeout reports that the gap was too long, then the stream continues and the late item arrives next.

21. Fallible streams: `TryStreamExt`

A stream of Results is a TryStream. TryStreamExt adds methods that work on the Ok values and stop at the first Err: map_ok, try_filter, and_then, and try_collect, which gives Result<Vec<T>, E>.

Rust

use futures::TryStreamExt;
use futures::stream;

async fn parse_line(line: &str) -> Result<(String, u32), String> {
    let (name, qty) = line.split_once(',').ok_or(format!("no comma in {line:?}"))?;
    let qty = qty.trim().parse().map_err(|_| format!("bad quantity in {line:?}"))?;
    Ok((name.to_string(), qty))
}

async fn load(lines: Vec<&str>) -> Result<Vec<String>, String> {
    stream::iter(lines.into_iter().map(Ok::<&str, String>)) // a stream of Result
        .and_then(parse_line)
        .try_filter(|(_, qty)| std::future::ready(*qty > 0))
        .map_ok(|(name, qty)| format!("{qty} x {name}"))
        .try_collect()
        .await
}

#[tokio::main]
async fn main() {
    println!("{:?}", load(vec!["pen, 3", "ink, 0", "paper, 10"]).await);
    println!("{:?}", load(vec!["pen, 3", "ink; 2", "paper, 10"]).await);
}

Output

Ok(["3 x pen", "10 x paper"])
Err("no comma in \"ink; 2\"")

try_collect stops at the first Err: in the second call, paper is never parsed.

22. Pinning: when `next()` needs `pin!`

next() requires the stream to be Unpin (safe to move while being polled). Streams built from async blocks, like unfold or then, usually aren't, and calling next() on them directly fails:

Rust✗ does not compile · E0277

use futures::StreamExt;
use futures::stream;

#[tokio::main]
async fn main() {
    let mut countdown = stream::unfold(3, |n| async move { if n == 0 { None } else { Some((n, n - 1)) } });
    while let Some(n) = countdown.next().await {
        println!("{n}");
    }
}

The compiler's message is error[E0277]: `{async block}` cannot be unpinned, followed by a long chain of notes ending in "consider using the pin! macro".

Pin it first: std::pin::pin! (or tokio::pin!) pins it on the stack, Box::pin on the heap. Consuming adapters like collect, fold, and for_each take the stream by value and don't need this.

Rust

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

#[tokio::main]
async fn main() {
    let mut countdown = pin!(stream::unfold(3, |n| async move { if n == 0 { None } else { Some((n, n - 1)) } }));
    while let Some(n) = countdown.next().await {
        println!("{n}...");
    }
    println!("liftoff");
}

Output

3...
2...
1...
liftoff

23. Returning streams from functions

Return impl Stream<Item = T> for a single concrete stream type. When different branches build different streams, box them into one type: BoxStream<'a, T> from futures (with the .boxed() method).

Rust

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

fn numbers(kind: &str) -> BoxStream<'static, u32> {
    match kind {
        "evens" => stream::iter((0..10).filter(|n| n % 2 == 0)).boxed(),
        "squares" => stream::iter(1..=4).map(|n| n * n).boxed(),
        _ => stream::empty().boxed(),
    }
}

#[tokio::main]
async fn main() {
    for kind in ["evens", "squares", "other"] {
        let v: Vec<u32> = numbers(kind).collect().await;
        println!("{kind}: {v:?}");
    }
}

Output

evens: [0, 2, 4, 6, 8]
squares: [1, 4, 9, 16]
other: []

A BoxStream is Unpin, so next() works on it directly.

24. Implementing `Stream` by hand

Like a hand-written Future, a hand-written Stream implements one method, poll_next, which returns Poll::Ready(Some(item)), Poll::Ready(None) at the end, or Poll::Pending. This one yields every line of a text, a little like BufReader::lines.

Rust

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

struct Lines {
    text: String,
    pos: usize,
}

impl Stream for Lines {
    type Item = String;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<String>> {
        if self.pos >= self.text.len() {
            return Poll::Ready(None);
        }
        let rest = &self.text[self.pos..];
        let end = rest.find('\n').map_or(rest.len(), |i| i);
        let line = rest[..end].to_string();
        self.pos += end + 1;
        Poll::Ready(Some(line)) // always ready: the data is in memory
    }
}

#[tokio::main]
async fn main() {
    let lines = Lines { text: String::from("GET /\nPOST /login\nGET /cart"), pos: 0 };
    let posts: Vec<String> = lines.filter(|l| std::future::ready(l.starts_with("POST"))).collect().await;
    println!("{posts:?}");
}

Output

["POST /login"]

In practice you rarely write poll_next yourself: unfold, channels with ReceiverStream, and adapters cover most needs.

25. Realistic: a paginated search with early exit

A search API returns results page by page. The caller wants the first 5 results with a score above a threshold. Thanks to laziness, the pipeline stops fetching as soon as it has 5.

Rust

use futures::StreamExt;
use futures::stream;
use std::future::ready;
use std::sync::atomic::{AtomicU32, Ordering};

static REQUESTS: AtomicU32 = AtomicU32::new(0);

#[derive(Debug)]
struct Hit {
    doc: String,
    score: u32,
}

async fn search_page(query: &str, page: u32) -> Option<Vec<Hit>> {
    REQUESTS.fetch_add(1, Ordering::Relaxed);
    tokio::task::yield_now().await;
    if page >= 10 {
        return None; // no more pages
    }
    Some((0..4).map(|i| Hit { doc: format!("{query}-p{page}-{i}"), score: (page * 7 + i * 13) % 100 }).collect())
}

#[tokio::main]
async fn main() {
    let hits = stream::unfold(0, |page| async move { search_page("rust", page).await.map(|hits| (hits, page + 1)) })
        .flat_map(stream::iter)
        .filter(|h| ready(h.score >= 50))
        .take(5);
    let hits: Vec<Hit> = hits.collect().await;

    for h in &hits {
        println!("{:<12} score {}", h.doc, h.score);
    }
    println!("requests made: {} of 10 pages", REQUESTS.load(Ordering::Relaxed));
}

Output

rust-p2-3    score 53
rust-p3-3    score 60
rust-p4-2    score 54
rust-p4-3    score 67
rust-p5-2    score 61
requests made: 6 of 10 pages

26. Realistic: downloading with bounded concurrency and error reporting

Download a list of URLs, at most 3 at a time, collecting successes and failures separately. buffer_unordered handles the concurrency; errors are just values in the stream.

Rust

use futures::StreamExt;
use futures::stream;
use std::time::Duration;

async fn download(url: &'static str) -> Result<(String, usize), String> {
    tokio::time::sleep(Duration::from_millis(url.len() as u64 * 3)).await;
    if url.contains("broken") {
        Err(format!("{url}: 404"))
    } else {
        Ok((url.to_string(), url.len() * 1024))
    }
}

#[tokio::main]
async fn main() {
    let urls = ["/a.png", "/broken.css", "/app.js", "/logo.svg", "/broken-link", "/index.html"];

    let results: Vec<Result<(String, usize), String>> = stream::iter(urls).map(download).buffer_unordered(3).collect().await;

    let (ok, failed): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_ok);
    let mut ok: Vec<(String, usize)> = ok.into_iter().map(Result::unwrap).collect();
    ok.sort();
    let mut failed: Vec<String> = failed.into_iter().map(Result::unwrap_err).collect();
    failed.sort();

    let total: usize = ok.iter().map(|(_, bytes)| bytes).sum();
    println!("downloaded {} files, {} KiB", ok.len(), total / 1024);
    println!("failed: {failed:?}");
}

Output

downloaded 4 files, 33 KiB
failed: ["/broken-link: 404", "/broken.css: 404"]

27. Realistic: a sensor pipeline

Two sensors feed readings through channels. The pipeline merges them, drops invalid readings, groups them into windows of 4, and computes an average per window.

Rust

use futures::StreamExt;
use futures::stream::select;
use std::future::ready;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

fn spawn_sensor(name: &'static str, readings: Vec<f64>) -> ReceiverStream<(&'static str, f64)> {
    let (tx, rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for r in readings {
            tx.send((name, r)).await.unwrap();
            tokio::task::yield_now().await;
        }
    });
    ReceiverStream::new(rx)
}

#[tokio::main]
async fn main() {
    let a = spawn_sensor("north", vec![20.0, 21.0, -999.0, 22.0, 23.0]);
    let b = spawn_sensor("south", vec![19.0, 18.0, 17.0, -999.0]);

    let windows: Vec<(usize, f64)> = select(a, b)
        .filter(|(_, v)| ready(*v > -100.0)) // -999 means "sensor error"
        .map(|(_, v)| v)
        .chunks(4)
        .map(|w| (w.len(), w.iter().sum::<f64>() / w.len() as f64))
        .collect()
        .await;

    let readings: usize = windows.iter().map(|(n, _)| n).sum();
    let overall = windows.iter().map(|(n, avg)| *n as f64 * avg).sum::<f64>() / readings as f64;
    println!("{} windows, {readings} valid readings", windows.len());
    println!("overall average: {overall:.2}");
}

Output

2 windows, 7 valid readings
overall average: 20.00

Individual window averages depend on how the two sensors' readings interleave, so this prints only order-independent results.

28. Iterators, streams, and futures side by side

	one value	many values
sync	`T`	`Iterator<Item = T>`
async	`Future<Output = T>`	`Stream<Item = T>`

Task	Iterator	Stream
get next	`it.next()`	`s.next().await`
transform	`map`	`map` (sync) / `then` (async)
filter	`filter(	x
reduce	`fold`	`fold` with an async closure
collect	`collect()`	`collect().await`
run many at once	`rayon`'s `par_iter`	`buffered(n)` / `buffer_unordered(n)`

Summary

A Stream is an async Iterator: next().await yields Some(item) or None. Use futures::StreamExt (plus tokio_stream::StreamExt for time-based adapters).
Create streams with stream::iter, unfold (stateful producers like pagination), once, repeat_with, ReceiverStream, and IntervalStream.
Adapters mirror Iterator: map, filter, take, fold, collect. Async steps use then, filter_map, for_each.
Streams are lazy and pull-based, so take stops upstream work and slow consumers get backpressure automatically.
buffered(n) (ordered) and buffer_unordered(n) (as-completed) run futures concurrently with a limit; for_each_concurrent when you need no results.
chunks, chunks_timeout, select, zip, take_until, timeout, and throttle handle batching, merging, and time.
TryStreamExt works on streams of Result. Pin streams with pin! before calling next() when they aren't Unpin; return impl Stream or BoxStream.

Next: How Async Works: Futures, Pin & Wakers, what happens under async and .await.

1. Consuming a stream

2. Adapters look just like Iterator

3. then: an async map

4. filter_map with async work

5. fold and for_each

6. Streams are lazy and pull-based

7. unfold for pagination

8. Stopping early saves work

9. Other ways to create streams

10. A channel as a stream

11. A timer as a stream

12. buffered: concurrent, in order

13. buffer_unordered: concurrent, as they finish

14. Limiting concurrency with buffer_unordered

15. for_each_concurrent

16. Batching with chunks

17. chunks_timeout: batch by size or time

18. Merging streams

19. take_while, skip_while, and take_until

20. Timeouts and throttling

21. Fallible streams: TryStreamExt

22. Pinning: when next() needs pin!

23. Returning streams from functions

24. Implementing Stream by hand

25. Realistic: a paginated search with early exit

26. Realistic: downloading with bounded concurrency and error reporting

27. Realistic: a sensor pipeline

28. Iterators, streams, and futures side by side

Summary

2. Adapters look just like `Iterator`

3. `then`: an async map

4. `filter_map` with async work

5. `fold` and `for_each`

7. `unfold` for pagination

12. `buffered`: concurrent, in order

13. `buffer_unordered`: concurrent, as they finish

14. Limiting concurrency with `buffer_unordered`

15. `for_each_concurrent`

16. Batching with `chunks`

17. `chunks_timeout`: batch by size or time

19. `take_while`, `skip_while`, and `take_until`

21. Fallible streams: `TryStreamExt`

22. Pinning: when `next()` needs `pin!`

24. Implementing `Stream` by hand