#async #non-blocking #futures #stream #iterator

stream-flatten-iters

Flattens a stream of iterators into one continuous stream

3 unstable releases

0.2.0 May 13, 2020
0.1.1 May 1, 2020
0.1.0 May 1, 2020

#46 in #non-blocking

Download history 14/week @ 2021-02-20 8/week @ 2021-02-27 14/week @ 2021-03-20 6/week @ 2021-03-27 12/week @ 2021-04-03 41/week @ 2021-04-10 26/week @ 2021-04-17 53/week @ 2021-04-24 41/week @ 2021-05-01 80/week @ 2021-05-08 56/week @ 2021-05-15 48/week @ 2021-05-22 17/week @ 2021-05-29 25/week @ 2021-06-05

108 downloads per month

MIT license

19KB
281 lines

flatten_iters flattens a stream of iterators into one continuous stream.

This is useful when you have a producer that is paging through a resource (like a REST endpoint with pages or a next URL, an ElasticSearch query with a scroll parameter, etc.)

This code is taken almost verbatim from StreamExt::flatten and is similar in spirit to Iterator::flatten.

use stream_flatten_iters::StreamExt as _;
use futures::stream::StreamExt;

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

    tokio::spawn(async move {
        tx.send(vec![0, 1, 2, 3]).await.unwrap();
        tx.send(vec![4, 5, 6]).await.unwrap();
        tx.send(vec![7, 8, 9]).await.unwrap();
    });

    let mut stream = rx.flatten_iters();

    while let Some(res) = stream.next().await {
        println!("got = {}", res);
    }
}

// Output:
// got = 0
// got = 1
// got = 2
// got = 3
// got = 4
// got = 5
// got = 6
// got = 7
// got = 8
// got = 9

This is especially useful when combined with StreamExt::buffered to keep a buffer of promises going throughout a long promise.

use stream_flatten_iters::StreamExt as _;
use futures::stream::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (mut tx, mut rx) = tokio::sync::mpsc::channel(3);

    tokio::spawn(async move {
        for i in 0_usize..100 {
            let start = i * 10;
            let end = start + 10;
            tx.send(start..end).await.unwrap();
        }
    });

    let mut stream = rx.flatten_iters().map(|i| long_process(i)).buffered(10);

    let mut total = 0_usize;
    while let Some(res) = stream.next().await {
        let _ = res?;
        total += 1;
        println!("Completed {} tasks", total);
    }

    Ok(())
}

async fn long_process(i: usize) -> Result<(), Box<dyn std::error::Error>> {
    // Do something that takes a long time
    Ok(())
}

Dependencies

~0.5–1MB
~21K SLoC