wingfoil 3.0.1

graph based stream processing framework
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

use async_stream::stream;
use futures::StreamExt;
use std::boxed::Box;
use std::pin::Pin;
use std::time::Duration;
use wingfoil::*;

fn main() {
    env_logger::init();
    let period = Duration::from_millis(10);
    let run_for = RunFor::Duration(period * 5);
    let run_mode = RunMode::RealTime;

    let producer = move |_| async move {
        Ok(stream! {
            for i in 0.. {
                tokio::time::sleep(period).await; // simulate waiting IO
                let time = NanoTime::now();
                yield Ok((time, i * 10));
            }
        })
    };

    let consumer = async move |mut source: Pin<Box<dyn FutStream<u32>>>| {
        while let Some((time, value)) = source.next().await {
            println!("{time:?}, {value:?}");
        }
        Ok(())
    };

    produce_async(producer)
        .logged("on-graph", log::Level::Info)
        .collapse()
        .consume_async(Box::new(consumer))
        .run(run_mode, run_for)
        .unwrap();
}