🎓 A1Learner
🔍

Rust Merging Streams

Merging streams means polling several streams together and taking each item as it becomes ready. The merged stream keeps every item from every source.

Two streams, one reader

stream::select puts two streams behind one name. next() takes the next ready item from either side.

let mut both = stream::select(stream::iter(vec!["a1", "a2"]), stream::iter(vec!["b1"]));
println!("first = {:?}", both.next().await);

Ready items take turns

Both streams here are always ready. select shares the turns fairly. The items alternate.

let merged: Vec<&str> = stream::select(stream::iter(vec!["a1", "a2"]), stream::iter(vec!["b1", "b2"]))
    .collect()
    .await;
println!("merged = {merged:?}");

A finished stream retires quietly

The short stream holds one item. It ends after that item. The long stream flows on alone.

let all: Vec<&str> = stream::select(stream::iter(vec!["only"]), stream::iter(vec!["x", "y", "z"]))
    .collect()
    .await;
println!("all = {all:?}");

Many streams at once

select_all merges a whole Vec of streams. Every stream must yield the same item type.

let streams = vec![stream::iter(vec![1, 4]), stream::iter(vec![2]), stream::iter(vec![3])];
let every: Vec<i32> = stream::select_all(streams).collect().await;
println!("every = {every:?}");

The tokio-stream spelling

Tokio code often uses merge instead. It does the same job. It lives on tokio-stream's StreamExt.

let merged: Vec<i32> = tokio_stream::iter(vec![1, 3])
    .merge(tokio_stream::iter(vec![2]))
    .collect()
    .await;
println!("merged = {merged:?}");

Example

// Cargo.toml: tokio = { version = "1", features = ["full"] }, futures = "0.3"
use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let fast = Box::pin(stream::iter(vec![("fast 1", 10), ("fast 2", 10), ("fast 3", 10)])
        .then(|(name, ms)| async move {
            sleep(Duration::from_millis(ms)).await;
            name
        }));
    let slow = Box::pin(stream::iter(vec![("slow 1", 45)]).then(|(name, ms)| async move {
        sleep(Duration::from_millis(ms)).await;
        name
    }));

    let mut both = stream::select(fast, slow);
    while let Some(name) = both.next().await {
        println!("{name}");
    }
    println!("both streams done");
}

Key points

  • Use stream::select(a, b) to merge exactly two streams into one
  • Use stream::select_all when the streams sit in a Vec
  • A finished stream retires on its own: the merged stream ends only when every side has ended
  • Items keep their own stream's order.
  • The mix between streams follows readiness, not a promise
  • Pin a merged stream's sources when they cannot be unpinned: Box::pin is the usual tool
  • stream::select merges streams and keeps every item.
  • Reading one stream to its end before the other starts is not merging.

In this example, two sensors report at different speeds. The fast one reports every 10 milliseconds. The slow one needs 45. Both streams sit in boxes. Their item futures cannot be unpinned. select needs Unpin sources, so Box::pin is the standard fix. The merge takes items as they land. The three fast reports arrive first. The slow report lands last. Then both streams end, and the loop closes.

Note

select! races futures and keeps one winner.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

You merge a chat stream with a timer stream. The chat stream ends. What happens to the timer items?