🎓 A1Learner
🔍

Rust Stream Adapters in Practice

A stream adapter is a method that takes a stream and returns a changed stream. Most adapters copy iterator adapters, name for name.

skip drops the first items

skip(2) throws the first two items away. The stream starts at the third.

let mut s = stream::iter(vec![10, 20, 30, 40]).skip(2);
println!("third = {:?}", s.next().await);

take_while ends the stream early

take_while tests each item. The first failure ends the stream for good. The 8 after the 7 never flows.

let kept: Vec<i32> = stream::iter(vec![2, 4, 6, 7, 8])
    .take_while(|n| future::ready(*n % 2 == 0))
    .collect()
    .await;
println!("kept = {kept:?}");

filter_map changes and drops in one step

The closure returns a future of an Option. Some keeps the new value. None drops the item.

let words = vec!["10", "oops", "30"];
let nums: Vec<i32> = stream::iter(words)
    .filter_map(|w| async move { w.parse::<i32>().ok() })
    .collect()
    .await;
println!("nums = {nums:?}");

then is map with an await inside

Each item waits for its future before the next one starts. Order survives. Speed does not multiply.

let doubled: Vec<i32> = stream::iter(vec![1, 2, 3])
    .then(|n| async move { n * 2 })
    .collect()
    .await;
println!("doubled = {doubled:?}");

fold rolls the stream into one value

fold is a consumer, like collect. It pulls every item. It hands back one running total.

let total: i32 = stream::iter(vec![1, 2, 3, 4])
    .fold(0, |acc, n| async move { acc + n })
    .await;
println!("total = {total}");

Example

// Cargo.toml: futures = "0.3"
use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

async fn run() {
    let readings = vec!["12", "skip-me", "7", "20", "3"];

    let total: i32 = stream::iter(readings)
        .filter_map(|w| async move { w.parse::<i32>().ok() })
        .take_while(|n| future::ready(*n < 20))
        .fold(0, |acc, n| async move { acc + n })
        .await;

    println!("total = {total}");
}

fn main() {
    block_on(run());
}

Key points

  • Use skip(n) to drop the first n items of a stream
  • Use take_while to end a stream at the first item that fails the test
  • Use filter_map when each item may change, vanish, or both
  • Use then when the change itself needs an await: one item at a time, in order
  • fold and collect are consumers: they pull the stream to its end and return one value
  • An adapter reads one stream and produces another stream from it.
  • The futures StreamExt holds the big catalogue.
  • tokio-stream StreamExt holds a smaller set.
  • A few adapters have no iterator twin at all.

In this example, five readings arrive as text. filter_map parses each one. The word "skip-me" fails to parse. It becomes None and drops out. take_while then keeps numbers under 20. The 20 stops the stream. The 3 behind it never arrives. fold adds what survives: 12 + 7 = 19.

Note

tokio-stream StreamExt adds time tools instead, like timeout and throttle.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

take_while meets its first failing item. Later items would pass the test again. What flows after the failure?