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
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

## async / tokio integration

In this example, we demonstrate wingfoil's integration with tokio / async.
This makes building IO adapters at the graph edge a breeze, giving 
the best of sync and async worlds.   

Async streams are a useful abstraction for IO but are not as powerful or 
as easy to use for implementing complex business logic as wingfoil.  Async 
uses an implicit, depth first graph execution strategy.  In
contrast wingfoil's explicit, breadth first graph execution algorithm, 
is easier to reason about, encourages re-use and provides a more 
structured and productive development environment.   Wingfoil's
explicit modelling of time with support for historical and realtime 
modes, is also huge win in terms of productivity and ability to 
backtest strategies over async streams.


The key methods here are produce_async which maps an async futures stream 
into a wingfoil stream and consume_async which takes a wingfoil stream
and consumes it with a supplied async closure.

This hybrid sync-async approach enforces clear separation 
between IO and business logic, which can often be problematic in 
async oriented systems.   


```bash
RUST_LOG=INFO cargo run --example async
```

```rust
use async_stream::stream;
use std::time::Duration;
use std::pin::Pin;
use futures::StreamExt;
use wingfoil::*;

let period = Duration::from_millis(10);
let run_for = RunFor::Duration(period * 5);
let run_mode = RunMode::RealTime;
let producer = move |_ctx: RunParams| 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);
```