stream.rs

use std::pin::pin;
use std::time::{Duration, Instant};

use async_io::{block_on, Timer};
use async_stream::stream;
use futures::StreamExt;
use futures_lite::future::yield_now;

async fn item() -> Duration {
    let start = Instant::now();
    yield_now().await;
    Instant::now().duration_since(start)
}

fn main() {
    let factory = stream! {
        loop {
            yield item();
        }
    }
    .buffered(5);

    let mut stream = pin!(factory);

    let result = block_on(async {
        let mut result = vec![];
        for _ in 0..5 {
            result.push(stream.next().await.unwrap());
            Timer::after(Duration::from_millis(500)).await;
        }
        result
    });
    dbg!(result);
    // [examples/buffered_stream.rs:35:5] result = [
    //     612.875µs,
    //     501.832917ms,
    //     1.002531209s,
    //     1.503673417s,
    //     2.004864417s,  <---- what the fuck?
    // ]
}