iterstream 0.1.2

Converts Iterator into real asynchronous Stream
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

use futures::StreamExt;
use futures::executor::ThreadPool;
use iterstream::IterStream;
use std::thread::sleep;
use std::time::{Duration, Instant};

// This struct will implement an iterator that blocks during 1 second
// before yielding each value

struct Sleep {
    max: u32,
    name: String,
}

impl Sleep {
    fn new(max: u32, name: &str) -> Self {
        Sleep {
            max,
            name: name.into(),
        }
    }
}

impl Iterator for Sleep {
    type Item = (u32, String);
    fn next(&mut self) -> Option<Self::Item> {
        match self.max {
            0 => None,
            _ => {
                sleep(Duration::from_secs(1));
                self.max -= 1;
                Some((self.max, self.name.clone()))
            }
        }
    }
}

// When calling next on 4 sleep iterators using a round robin
// scheduling, we block for one seconds at each next() call. Thus,
// the total execution time of this function is 20 seconds

fn round_robin_iterator() {
    let mut iterators = vec![
        Sleep::new(5, "one"),
        Sleep::new(5, "two"),
        Sleep::new(5, "three"),
        Sleep::new(5, "four"),
    ];

    loop {
        let mut one_yield = false;
        for sleep in iterators.iter_mut() {
            if let Some((val, name)) = sleep.next() {
                println!("{} {}", name, val);
                one_yield = true;
            }
        }
        if !one_yield {
            break;
        }
    }
}

// When doin the same thing using IterStream, four tasks are created,
// one for each Sleep Iterator. Those tasks call next() on the
// iterator asynchronously so every second, the 4 streams can yield a
// new value immediatly, leading to around 5 seconds execution time
// for this function that does exactly the same thing as the previous
// one.
async fn round_robin_streams() {
    // Create a thread pool for the tasks
    let pool = ThreadPool::new().unwrap();
    let mut iterators = vec![
        // cloning the pool only clones the internal reference to the
        // pool, it does not create a new pool
        Sleep::new(5, "one").to_stream_with_pool(20, pool.clone()), 
        Sleep::new(5, "two").to_stream_with_pool(20, pool.clone()),
        Sleep::new(5, "three").to_stream_with_pool(20, pool.clone()),
        Sleep::new(5, "four").to_stream_with_pool(20, pool.clone()),
    ];
    loop {
        let mut one_yield = false;
        for sleep in iterators.iter_mut() {
            if let Some((val, name)) = sleep.next().await {
                println!("{} {}", name, val);
                one_yield = true;
            }
        }
        if !one_yield {
            break;
        }
    }
}

fn main() {
    println!("With iterators");
    let start = Instant::now();
    round_robin_iterator();
    println!("Duration: {} s", start.elapsed().as_secs_f32());
    println!("With streams");
    let start = Instant::now();
    futures::executor::block_on(async {
        round_robin_streams().await;
    });
    println!("Duration: {} s", start.elapsed().as_secs_f32());
}