Wingfoil

https://docs.rs/wingfoil/4.0.1
Admin
Wingfoil is a blazingly fast, highly scalable stream processing framework designed for...
Tokens:168,906
Snippets:1,480
Trust Score:9.1
Update:5 months ago
Show doc for...
Tokens:
Raw
### Node Setup and Start

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Initializes and starts the nodes within the graph. This is a prerequisite for the main execution loop.

```rust
pub fn run(&mut self) -> anyhow::Result<()> {
        self.setup_nodes()?;
        self.start_nodes()?;
```

--------------------------------

### CallBackStream setup method

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.CallBackStream.html

Called by the graph after wiring and before the start method. Useful for any setup logic that needs to occur after wiring is complete.

```rust
fn setup(&mut self, state: &mut GraphState) -> Result<()>
```

--------------------------------

### Quick Start: Linear Pipeline

Source: https://docs.rs/wingfoil/4.0.1/index.html

Build a simple, linear data processing pipeline. This example demonstrates a basic ticker, counter, formatter, and printer.

```rust
use wingfoil::*
use std::time::Duration;
fn main() {
    let period = Duration::from_secs(1);
    ticker(period)
        .count()
        .map(|i| format!("hello, world {:}", i))
        .print()
        .run(RunMode::RealTime, RunFor::Duration(period*3)
    );
}
```

--------------------------------

### Node Start

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Applies the start function to all active nodes in the graph. This is typically called after setup to begin node execution.

```APIDOC
## Graph::start_nodes

### Description
Applies the `start` method to all active nodes within the graph.

### Method
`start_nodes(&mut self) -> anyhow::Result<()>`

### Returns
`Ok(())` if all nodes were started successfully, otherwise an `anyhow::Result` containing an error.
```

--------------------------------

### Node Setup

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Applies the setup function to all active nodes in the graph. This is typically called during the graph's initialization phase.

```APIDOC
## Graph::setup_nodes

### Description
Applies the `setup` method to all active nodes within the graph.

### Method
`setup_nodes(&mut self) -> anyhow::Result<()>`

### Returns
`Ok(())` if all nodes were set up successfully, otherwise an `anyhow::Result` containing an error.
```

--------------------------------

### etcd Example: Watch, Transform, and Write

Source: https://docs.rs/wingfoil/4.0.1/index.html

Watch a key prefix in etcd, transform the values, and write the results back. This example demonstrates a round-trip data flow.

```rust
use wingfoil::adapters::etcd::*;
use wingfoil::*;

let conn = EtcdConnection::new("http://localhost:2379");

let round_trip = etcd_sub(conn.clone(), "/source/")
    .map(|burst| {
        burst.into_iter().map(|event| {
            let upper = event.entry.value_str().unwrap_or("").to_uppercase().into_bytes();
            EtcdEntry { key: event.entry.key.replacen("/source/", "/dest/", 1), value: upper }
        })
        .collect::<Burst<EtcdEntry>>()
    })
    .etcd_pub(conn, None, true);

round_trip.run(RunMode::RealTime, RunFor::Cycles(3)).unwrap();
```

--------------------------------

### BenchTriggerNode start method implementation

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/bencher.rs.html

Illustrates the `start` method of `BenchTriggerNode`, which spawns a thread to monitor a shared signal and react to `Begin` and `Kill` states.

```rust
fn start(&mut self, state: &mut GraphState) {
            let signal = self.signal.clone();
            let notifier = state.ready_notifier();
            std::thread::spawn(move || {
                loop {
                    match signal.load(Ordering::SeqCst).into() {
                        Signal::Begin => {
                            signal.store(Signal::Running.into(), Ordering::SeqCst);
                            notifier.notify();
                        },
                        Signal::Kill => {
                            break;
                        },
                        _ => {
                            std::hint::spin_loop();
                        },
                    }
                }
            });
        }
```

--------------------------------

### produce_async Usage Example

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/fn.produce_async.html

An example demonstrating how to use the produce_async function with a closure that adapts its behavior based on context.

```APIDOC
produce_async(|ctx| async move {
    let start = ctx.start_time;
    let end = ctx.end_time();
    Ok(async_stream::stream! {
        // Use start, end in async code...
    })
})
```

--------------------------------

### Apply Setup Function to Nodes

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Applies a setup function to all active nodes in the graph. This is typically called once during graph initialization.

```rust
pub(crate) fn setup_nodes(&mut self) -> anyhow::Result<()> {
        self.apply_nodes("setup", |node, state| node.setup(state))
    }
```

--------------------------------

### Run a Node and Get Result

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/mod.rs.html

Initializes and executes a graph starting from the given node. Returns a result indicating success or failure.

```rust
/// Shortcut for [Graph::run] i.e. initialise and execute the graph.
/// ```
/// # use wingfoil::*
/// # use std::time::Duration;
/// let count = ticker(Duration::from_millis(1))
///     .count();
/// count.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Cycles(3))
///     .unwrap();
/// count.peek_value(); // 3
/// ```
fn run(self: &Rc<Self>, run_mode: RunMode, run_to: RunFor) -> anyhow::Result<()>;
fn into_graph(self: &Rc<Self>, run_mode: RunMode, run_for: RunFor) -> Graph;
```

```rust
fn run(self: &Rc<Self>, run_mode: RunMode, run_for: RunFor) -> anyhow::Result<()> {
    Graph::new(vec![self.clone()], run_mode, run_for).run()
}
fn into_graph(self: &Rc<Self>, run_mode: RunMode, run_for: RunFor) -> Graph {
    Graph::new(vec![self.clone()], run_mode, run_for)
}
```

--------------------------------

### Emit Tracing Events with Log Levels

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/lib.rs.html

This example demonstrates how to use the `logged` adapter to emit tracing events from a stream. It requires the `tracing` feature to be enabled and a tracing subscriber to be installed. The `run` method is called with specific historical and cycle configurations.

```rust
ticker(Duration::from_secs(1))
    .count()
    .logged("tick", Info)  // emits a tracing event per tick when feature = "tracing"
    .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Cycles(3))
    .unwrap();
```

--------------------------------

### Wingfoil Graph Execution Example

Source: https://docs.rs/wingfoil/4.0.1/wingfoil

Demonstrates basic graph execution in Wingfoil, showing how nodes with different frequencies are handled. This example processes a counter, filters for even and odd numbers, and prints them. The graph runs for a specified duration.

```rust
use wingfoil::*;
use std::time::Duration;

fn main() {
    let period = Duration::from_millis(10);
    let source = ticker(period).count(); // 1, 2, 3 etc
    let is_even = source.map(|i| i % 2 == 0);
    let odds = source
        .filter(is_even.not())
        .map(|i| format!({"i"});
    let evens = source
        .filter(is_even)
        .map(|i| format!({"i"});
    merge(vec![odds, evens])
        .print()
        .run(
            RunMode::HistoricalFrom(NanoTime::ZERO),
            RunFor::Duration(period * 5),
        );
}
```

--------------------------------

### Dynamic Graph Execution Commands

Source: https://docs.rs/wingfoil/4.0.1/index.html

Commands to run Wingfoil examples demonstrating dynamic graph modifications at runtime. These examples showcase different approaches to adding and removing nodes without stopping execution.

```bash
cargo run --example dynamic-group --features dynamic-graph-beta
```

```bash
cargo run --example dynamic-manual --features dynamic-graph-beta
```

```bash
cargo run --example demux
```

--------------------------------

### DemuxVecParent Setup Logic

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/demux.rs.html

Handles the setup phase for DemuxVecParent, resolving graph indices for child nodes and the overflow node, and ensuring that child nodes have been successfully registered with the graph.

```rust
fn setup(&mut self, graph_state: &mut GraphState) -> anyhow::Result<()> {
        let mut childes = self.children.borrow_mut();
        let mut node_indexes: Vec<_> = childes
            .drain( ..)
            .map(|strm| {
                graph_state.node_index(strm.as_node()).unwrap_or_else(|| {
                    panic!("Failed to resolve graph index of demux child node.  Was it added to the graph?")
                })
            })
            .collect();
        drop(childes);
        node_indexes
            .drain( ..)
            .for_each(|node_index| self.index_map.push(node_index));
        let overflow = self.overflow_child.take().unwrap();
        self.overflow_graph_index = graph_state.node_index(overflow.as_node());
        anyhow::ensure!(
            self.overflow_graph_index.is_some(),
            "Failed to resolve graph index of demux overflow node.  Was it added to the graph?"
        );
        anyhow::ensure!(
            !self.index_map.is_empty(),
            "Failed to resolve any children to demux into"
        );
        Ok(())
    }
```

--------------------------------

### ZeroMQ Publisher Example

Source: https://docs.rs/wingfoil/4.0.1/wingfoil

Publishes a stream of byte data over ZeroMQ on a specified port. Ensure the port is available and accessible.

```rust
use wingfoil::adapters::zmq::ZeroMqPub;
use wingfoil::*;

ticker(Duration::from_millis(100))
    .count()
    .map(|n: u64| format!({"n"}).into_bytes())
    .zmq_pub(7779, ())
    .run(RunMode::RealTime, RunFor::Forever)?;
```

--------------------------------

### Apply Start Function to Nodes

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Applies a start function to all active nodes in the graph. This is called to begin node execution.

```rust
pub(crate) fn start_nodes(&mut self) -> anyhow::Result<()> {
        self.apply_nodes("start", |node, state| node.start(state))
    }
```

--------------------------------

### Setup GraphProducerStream

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/graph_node.rs.html

Sets up the GraphProducerStream by executing the provided function in a new thread. It handles different run modes and creates a channel for communication.

```rust
let state = mem::take(&mut self.state);
        match state {
            GraphProducerStreamState::Func(func) => {
                let run_for = graph_state.run_for();
                let notifier = match graph_state.run_mode() {
                    RunMode::HistoricalFrom(_) => None,
                    RunMode::RealTime => Some(graph_state.ready_notifier()),
                };
                let (sender, receiver) = channel_pair(notifier);
                let mut receiver_stream = ChannelReceiverStream::new(receiver, None, None);
                receiver_stream.setup(graph_state)?;
                self.receiver_stream.set(receiver_stream).unwrap();
                let tokio_runtime = graph_state.tokio_runtime();
                let start_time = graph_state.start_time();
                let run_mode = graph_state.run_mode();
                let task = move || {
                    let node = func().send(sender, None);
                    let mut graph =
                        Graph::new_with(vec![node], tokio_runtime, run_mode, run_for, start_time);
                    graph.run().unwrap();
                };

                let handle = thread::spawn(task);
                self.state = GraphProducerStreamState::Handle(handle);
            }
            _ => panic!(),
        }
        Ok(())
```

--------------------------------

### Example Usage of feedback()

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/feedback.rs.html

This example demonstrates how to use the `feedback` function to create a feedback loop. It sets up a ticker, a feedback channel, and a `bimap` node that uses both the ticker's count and the feedback stream's previous value to calculate a sum. The result is then fed back into the channel using `feedback`.

```rust
# use wingfoil::*
# use std::time::Duration;
let src = ticker(Duration::from_nanos(100)).count();
let (tx, rx) = feedback::<u64>();

let sum = bimap(
    Dep::Active(src.clone()),
    Dep::Passive(rx),
    |count, prev| count + prev,
);

let writer = sum.feedback(tx);
// writer is Rc<dyn Stream<u64>> — values pass through, feedback is a side effect

```

--------------------------------

### Async IO Example with Real-Time and Historical Modes

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/async_io.rs.html

Demonstrates a complete asynchronous I/O pipeline with both real-time and historical run modes. It includes a producer that generates data and a consumer that processes it.

```rust
let example_producer = move |ctx: RunParams| async move {
    Ok(async_stream::stream! {
        for i in 0.. {
            let time = match ctx.run_mode {
                RunMode::HistoricalFrom(_) => {
                    // wire up historical source here
                    ctx.start_time + period * i
                },
                RunMode::RealTime => {
                    // wire up real time source here
                    tokio::time::sleep(period).await; // simulate waiting IO
                    NanoTime::now()
                }
            };
            yield Ok((time, i * 10));
        }
    })
};

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

produce_async(example_producer)
    .collapse()
    .logged("on-graph", log::Level::Info)
    .consume_async(Box::new(example_consumer))
    .run(run_mode, run_for)
    .unwrap();
```

--------------------------------

### Graph Execution with DoublerNode in Rust

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

This snippet sets up a graph with a DoublerNode and a CountedNode, runs it for a few cycles, and asserts that setup and start methods are called only once.

```rust
let setup_count = Rc::new(RefCell::new(0u32));
let start_count = Rc::new(RefCell::new(0u32));
let ticker = ticker(Duration::from_nanos(1));
let counted = Rc::new(RefCell::new(CountedNode {
    ticker: ticker.clone(),
    setup_count: setup_count.clone(),
    start_count: start_count.clone(),
}));
let doubler = Rc::new(RefCell::new(DoublerNode {
    trigger: ticker.clone(),
    counted: counted.clone().as_node(),
    add_count: 0,
}));

Graph::new(
    vec![doubler.clone().as_node()],
    RunMode::HistoricalFrom(NanoTime::ZERO),
    RunFor::Cycles(3),
)
.run()
.unwrap();

// setup and start should only be called once even though add_upstream was called twice
assert_eq!(*setup_count.borrow(), 1, "setup called once");
assert_eq!(*start_count.borrow(), 1, "start called once");
```

--------------------------------

### Emit Tracing Events with Ticker

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/index.html

This example demonstrates how to use the `ticker` and `logged` functions to emit tracing events. Ensure the `tracing` feature flag is enabled and a tracing subscriber is installed. The `logged` function emits a tracing event per tick when the `tracing` feature is active.

```rust
use log::Level::Info;
use std::time::Duration;
use wingfoil::*;

// With the `tracing` feature and a subscriber installed:
// tracing_subscriber::fmt::init();

ticker(Duration::from_secs(1))
    .count()
    .logged("tick", Info)  // emits a tracing event per tick when feature = "tracing"
    .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Cycles(3))
    .unwrap();

```

--------------------------------

### ZeroMQ Publisher Example

Source: https://docs.rs/wingfoil/4.0.1/index.html

Publishes a stream of byte data over ZeroMQ on a specified port. This can be subscribed to by other processes, including Python clients.

```rust
// publisher
use wingfoil::adapters::zmq::ZeroMqPub;
use wingfoil::*;

ticker(Duration::from_millis(100))
    .count()
    .map(|n: u64| format!({"n"}).into_bytes())
    .zmq_pub(7779, ())
    .run(RunMode::RealTime, RunFor::Forever)?;

```

--------------------------------

### Wingfoil Graph Execution Example

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/lib.rs.html

Demonstrates how to build and run a simple graph in Wingfoil. Use this to understand basic node creation, mapping, filtering, and merging operations.

```rust
#![warn(clippy::perf)]
#![allow(clippy::type_complexity)]
#![allow(clippy::needless_doctest_main)]
#![doc = include_str!("../README.md")]

//! ## Graph Execution

//! Wingfoil abstracts away the details of how to co-ordinate the calculation of your
//! application, parts of which may be executing at different frequencies.

//! Only the nodes that actually require cycling are executed which allows wingfoil to
//! efficiently scale to very large graphs.

//! Consider the following example:
//! ```rust
//! use wingfoil::*
//! use std::time::Duration;

//! fn main() {
//!     let period = Duration::from_millis(10);
//!     let source = ticker(period).count(); // 1, 2, 3 etc
//!     let is_even = source.map(|i| i % 2 == 0);
//!     let odds = source
//!         .filter(is_even.not())
//!         .map(|i| format!({"%"}:} is odd", i));
//!     let evens = source
//!         .filter(is_even)
//!         .map(|i| format!({"%"}:} is even", i));
//!     merge(vec![odds, evens])
//!         .print()
//!         .run(
//!             RunMode::HistoricalFrom(NanoTime::ZERO),
//!             RunFor::Duration(period * 5),
//!         );
//! }
//! ```
//! This output is produced:  
//! ```pre
//! 1 is odd
//! 2 is even
//! 3 is odd
//! 4 is even
//! 5 is odd
//! 6 is even
//! ````
//! We can visualise the graph like this:
//! <div align="center">
//! <img alt="diagram" src="https://raw.githubusercontent.com/wingfoil-io/wingfoil/refs/heads/main/wingfoil/diagrams/odds_evens.png" width="400"/>
//! </div>
//! The input and output nodes tick 6 times each and the evens and odds nodes tick 3 times.

```

--------------------------------

### Node::start Method

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/trait.Node.html

Starts the node's operation. This method is called when the graph begins execution.

```rust
fn start(&self, state: &mut GraphState) -> Result<()>
```

--------------------------------

### CallBackStream start method

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.CallBackStream.html

Called by the graph after wiring and before the first cycle. Can be used to request an initial callback or set up initial state.

```rust
fn start(&mut self, state: &mut GraphState) -> Result<()>
```

--------------------------------

### Time and Timing Functions

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Provides access to the current engine time, elapsed time since start, and the graph's start time.

```APIDOC
## time()

### Description
Returns the current engine time.

### Method
`time()`

### Returns
- `NanoTime`: The current engine time.
```

```APIDOC
## elapsed()

### Description
Calculates and returns the time elapsed since the graph started.

### Method
`elapsed()`

### Returns
- `NanoTime`: The duration since the graph's start time.
```

```APIDOC
## start_time()

### Description
Returns the recorded start time of the graph execution.

### Method
`start_time()`

### Returns
- `NanoTime`: The timestamp when the graph execution began.
```

--------------------------------

### ZeroMQ Subscriber Example

Source: https://docs.rs/wingfoil/4.0.1/index.html

Subscribes to a ZeroMQ stream of byte data from a specified TCP address. It then prints the received data.

```rust
// subscriber
use wingfoil::adapters::zmq::zmq_sub;
use wingfoil::*;

let (data, _status) = zmq_sub::<Vec<u8>>("tcp://127.0.0.1:7779")?;
// See wingfoil-python/examples/zmq/ for a Python subscriber
data.print()
    .run(RunMode::RealTime, RunFor::Forever)?;

```

--------------------------------

### Graph::new_with

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.Graph.html

Creates a new Graph with additional configuration including a Tokio runtime, start time, and other parameters.

```APIDOC
## Graph::new_with

### Description
Creates a new Graph with additional configuration including a Tokio runtime, start time, and other parameters.

### Signature
```rust
pub fn new_with(root_nodes: Vec<Rc<dyn Node>>, tokio_runtime: Arc<Runtime>, run_mode: RunMode, run_for: RunFor, start_time: NanoTime) -> Graph
```

### Parameters
* `root_nodes` (Vec<Rc<dyn Node>>) - The root nodes for the graph.
* `tokio_runtime` (Arc<Runtime>) - The Tokio runtime to use for asynchronous operations.
* `run_mode` (RunMode) - The mode in which the graph should run.
* `run_for` (RunFor) - The duration for which the graph should run.
* `start_time` (NanoTime) - The specific start time for the graph execution.
```

--------------------------------

### Node Setup Logic

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/channel.rs.html

Configures the node based on the graph's run mode. For real-time mode, it sends a ready notifier. For historical mode, it sets up a callback if no trigger is present.

```rust
    fn setup(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
        match state.run_mode() {
            RunMode::RealTime => {
                if let Some(chan) = self.notifier_channel.take() {
                    chan.send(state.ready_notifier())
                        .map_err(|e| anyhow::anyhow!(e))?;
                }
            }
            RunMode::HistoricalFrom(time) => {
                if self.trigger.is_none() {
                    state.add_callback(time);
                }
            }
        }
        Ok(())
    }
```

--------------------------------

### Test `seen` Prevents Double Setup

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Ensures that the `setup` method is called only once for a node, even if it's added multiple times, by utilizing the `seen` mechanism.

```rust
            struct CountedNode {
                ticker: Rc<dyn Node>,
                setup_count: Rc<RefCell<u32>>,
                start_count: Rc<RefCell<u32>>,
            }

            impl MutableNode for CountedNode {
                fn upstreams(&self) -> UpStreams {
                    UpStreams::new(vec![self.ticker.clone()], vec![])
                }

                fn cycle(&mut self, _state: &mut GraphState) -> anyhow::Result<bool> {
                    Ok(true)
                }

                fn setup(&mut self, _state: &mut GraphState) -> anyhow::Result<()> {
                    *self.setup_count.borrow_mut() += 1;
                    Ok(())
                }

```

--------------------------------

### AsyncConsumerNode Setup Logic

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/async_io.rs.html

Initializes and spawns the asynchronous consumer task. It prepares the input stream and executes the provided future-based consumer function within a Tokio runtime.

```rust
fn setup(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
    let run_mode = state.run_mode();
    let run_for = state.run_for();
    let rx = self
        .rx
        .take()
        .ok_or_else(|| anyhow::anyhow!("rx is already taken"))?;

    let func = self
        .func
        .take()
        .ok_or_else(|| anyhow::anyhow!("func is already taken"))?;
    let ctx = RunParams {
        run_mode,
        run_for,
        start_time: state.start_time(),
    };

    let f = async move {
        let src = rx
            .to_boxed_message_stream()
            .limit(run_mode, run_for)
            .to_stream();
        let fut = func(ctx, Box::pin(src));
        fut.await
    };

    let handle = state.tokio_runtime().spawn(f);

```

--------------------------------

### BFS vs DFS Benchmark Output

Source: https://docs.rs/wingfoil/4.0.1/index.html

Example output from the `bfs_vs_dfs` benchmarks, highlighting Wingfoil's flat performance compared to the exponential cost of depth-first approaches.

```text
1 ticks processed in 7.207µs, 7.207µs average.
value 170141183460469231731687303715884105728
```

--------------------------------

### ZeroMQ Subscriber Example

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/index.html

Subscribes to a stream of byte data from a ZeroMQ publisher. It connects to the specified TCP address and port, then processes and prints the received data.

```rust
use wingfoil::adapters::zmq::zmq_sub;
use wingfoil::*;

let (data, _status) = zmq_sub::<Vec<u8>>("tcp://127.0.0.1:7779")?;
// See wingfoil-python/examples/zmq/ for a Python subscriber
data.print()
    .run(RunMode::RealTime, RunFor::Forever?);
```

--------------------------------

### Wingfoil Breadth-First Graph Execution

Source: https://docs.rs/wingfoil/4.0.1/wingfoil

Illustrates Wingfoil's breadth-first graph execution, which avoids the exponential complexity of depth-first approaches. This example creates a deep graph of additions.

```rust
use wingfoil::*;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    env_logger::init();
    let mut source = constant(1_u128);
    for _ in 1..128 {
        source = add(&source, &source);
    }
    source
        .timed()
        .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)?;
    println!("value {{:?}}", source.peek_value());
    Ok(())
}
```

--------------------------------

### Accumulate Feedback Node Example

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/feedback.rs.html

Demonstrates accumulating values from a feedback node and asserting the final collected values. This is useful for testing data aggregation over time.

```rust
let res = value.accumulate().finally(|values, _| {
            assert_eq!(vec![1, 12, 123, 1234, 12345, 123456], values);
            Ok(())
        });

        Graph::new(
            vec![fb.as_node(), res],
            RunMode::HistoricalFrom(NanoTime::ZERO),
            RunFor::Duration(period * 5),
        )
        .run()
        .unwrap();
```

--------------------------------

### Node::setup Method

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/trait.Node.html

Initializes the node. This method is called once when the graph is set up.

```rust
fn setup(&self, state: &mut GraphState) -> Result<()>
```

--------------------------------

### Get elapsed time since start for a Stream

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/mod.rs.html

Calculates the time elapsed since the stream started. The stream must be run to produce a value.

```rust
let src: Rc<dyn Stream<u64>> = make_source(7, 50);
let te = src.ticked_at_elapsed();
te.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
    .unwrap();
assert_eq!(te.peek_value(), NanoTime::new(50));
```

--------------------------------

### Get elapsed time since start for a Node

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/mod.rs.html

Calculates the time elapsed since the stream started for a node. The stream must be run to produce a value.

```rust
let src: Rc<dyn Stream<u64>> = make_source(42, 100);
let node: Rc<dyn Node> = src.clone().as_node();
let te = node.ticked_at_elapsed();
te.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
    .unwrap();
// elapsed is time - start, start is ZERO, so elapsed == tick time
assert_eq!(te.peek_value(), NanoTime::new(100));
```

--------------------------------

### Create Graph with Async Runtime

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Creates a new Graph instance with an optional Tokio runtime for asynchronous operations. Allows specifying start time and run configuration.

```rust
pub fn new_with(
        root_nodes: Vec<Rc<dyn Node>>,
        tokio_runtime: Arc<tokio::runtime::Runtime>,
        run_mode: RunMode,
        run_for: RunFor,
        start_time: NanoTime,
    ) -> Graph {
        let state = GraphState::new(run_mode, run_for, start_time);
        state.run_time.set(tokio_runtime).ok();
        let mut graph = Graph { state };
        graph.initialise(root_nodes);
        graph
    }
```

--------------------------------

### KDB+ Example: Time-Sliced Queries

Source: https://docs.rs/wingfoil/4.0.1/index.html

Stream time-sliced queries directly into your graph from a KDB+ database. Requires a typed struct implementing KdbDeserialize.

```rust
#[derive(Debug, Clone, Default)]
struct Price {
    sym: Sym,
    mid: f64,
}

kdb_read::<Price, _>(
    conn,
    Duration::from_secs(10),
    |(t0, t1), _date, _iter| {
        format!(
            "select time, sym, mid from prices \ 
             where time >= (`timestamp$){{}}j, time < (`timestamp$){{}}j",
            t0.to_kdb_timestamp(),
            t1.to_kdb_timestamp(),
        )
    },
)
.logged("prices", Info)
.run(
    RunMode::HistoricalFrom(NanoTime::from_kdb_timestamp(0)),
    RunFor::Duration(Duration::from_secs(100)),
)?;
```

--------------------------------

### Bencher Struct and Methods

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/bencher.rs.html

Manages the lifecycle of a Wingfoil graph for benchmarking. Includes methods to create, start, step through, and stop the benchmark.

```rust
struct Bencher {
    run_mode: RunMode,
    signal: Arc<AtomicU8>,
    builder: Option<Box<dyn BenchBuilder>>,
}

impl Bencher {
    pub fn new<B: BenchBuilder + 'static>(builder: B, run_mode: RunMode) -> Self {
        let signal = Arc::new(AtomicU8::new(Signal::Ready.into()));
        let builder: Option<Box<dyn BenchBuilder>> = Some(Box::new(builder));
        Self {
            signal,
            builder,
            run_mode,
        }
    }

    pub fn step(&mut self) {
        self.signal.store(Signal::Begin.into(), Ordering::SeqCst);
        loop {
            match self.signal.load(Ordering::SeqCst).into() {
                Signal::End => {
                    break;
                }
                _ => {
                    std::hint::spin_loop();
                }
            }
        }
    }

    pub fn start(&mut self) {
        let builder = self.builder.take().unwrap();
        let signal = self.signal.clone();
        let run_mode = self.run_mode;
        std::thread::spawn(move || {
            let trigger = BenchTriggerNode::new(signal.clone()).into_node();
            let node = builder(trigger).produce(move || {
                signal.store(Signal::End.into(), Ordering::SeqCst);
            });
            Graph::new(vec![node], run_mode, RunFor::Forever)
                .run()
                .unwrap();
        });
    }

    pub fn stop(&self) {
        self.signal
            .clone()
            .store(Signal::Kill.into(), Ordering::SeqCst);
    }
}
```

--------------------------------

### CountedNode Implementation in Rust

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

A basic node that increments counters on setup and start. Useful for testing node lifecycle events.

```rust
struct CountedNode {
    ticker: Rc<dyn Node>,
    setup_count: Rc<RefCell<u32>>,
    start_count: Rc<RefCell<u32>>,
}

impl Node for CountedNode {
    fn setup(&self, _state: &mut GraphState) -> anyhow::Result<()> {
        *self.setup_count.borrow_mut() += 1;
        Ok(())
    }

    fn start(&mut self, _state: &mut GraphState) -> anyhow::Result<()> {
        *self.start_count.borrow_mut() += 1;
        Ok(())
    }
}
```

--------------------------------

### Compare RealTime vs Historical RunMode in Wingfoil

Source: https://docs.rs/wingfoil/4.0.1/wingfoil

This example demonstrates the difference in logging behavior between RealTime and Historical RunModes. RealTime is for production, while Historical is for testing and back-testing.

```Rust
use wingfoil::*;
use std::time::Duration;
use log::Level::Info;

pub fn main() {
    env_logger::init();
    for run_mode in vec![
        RunMode::RealTime,
        RunMode::HistoricalFrom(NanoTime::ZERO)
    ] {
        println!("\nUsing RunMode::{:?}", run_mode);
        ticker(Duration::from_secs(1))
            .count()
            .logged("tick", Info)
            .run(run_mode, RunFor::Cycles(3));
    }
}
```

--------------------------------

### Get Tick Timestamp (Elapsed)

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/mod.rs.html

Emits the time of source ticks relative to the start of the stream. Requires `wingfoil` and `std::time::Duration`.

```rust
/// Emits the time of source ticks relative to the start.
/// ```
/// # use wingfoil::*
/// # use std::time::Duration;
/// // 0, 1000000000, 2000000000, etc.
/// ticker(Duration::from_millis(10)).ticked_at_elapsed();
/// ```
#[must_use]
fn ticked_at_elapsed(self: &Rc<Self>) -> Rc<dyn Stream<NanoTime>>;
```

```rust
fn ticked_at_elapsed(self: &Rc<Self>) -> Rc<dyn Stream<NanoTime>> {
    let f = Box::new(|state: &mut GraphState| state.elapsed());
    GraphStateStream::new(self.clone(), f).into_stream()
}
```

--------------------------------

### Get Elapsed Tick Time with NodeOperators

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/trait.NodeOperators.html

Use `ticked_at_elapsed` to emit the time elapsed since the start of the stream for each tick. Useful for measuring durations.

```rust
// 0, 1000000000, 2000000000, etc.
ticker(Duration::from_millis(10)).ticked_at_elapsed();
```

--------------------------------

### MutableNode implementation for ChannelReceiverStream

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.ChannelReceiverStream.html

Implements the `MutableNode` trait for `ChannelReceiverStream`, defining methods for graph interaction such as `upstreams`, `cycle`, `setup`, `teardown`, `start`, `stop`, and `type_name`.

```rust
fn upstreams(&self) -> UpStreams
```

```rust
fn cycle(&mut self, state: &mut GraphState) -> Result<bool>
```

```rust
fn setup(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn teardown(&mut self, _: &mut GraphState) -> Result<()>
```

```rust
fn start(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn stop(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn type_name(&self) -> String
```

--------------------------------

### MutableNode Trait Implementation for SimpleIteratorStream

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.SimpleIteratorStream.html

Implements the MutableNode trait for SimpleIteratorStream, providing methods for graph interaction like cycle, start, upstreams, setup, stop, teardown, and type_name.

```rust
fn cycle(&mut self, state: &mut GraphState) -> Result<bool>
```

```rust
fn start(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn upstreams(&self) -> UpStreams
```

```rust
fn setup(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn stop(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn teardown(&mut self, state: &mut GraphState) -> Result<()>
```

```rust
fn type_name(&self) -> String
```

--------------------------------

### Emit Tracing Events with Wingfoil

Source: https://docs.rs/wingfoil/4.0.1/index.html

This example demonstrates how to emit tracing events per tick using the `logged` operator. Ensure the `tracing` feature flag is enabled and a tracing subscriber is installed. The `logged` operator emits events via the `log` crate, which are then handled by the `tracing` ecosystem.

```rust
use log::Level::Info;
use std::time::Duration;
use wingfoil::*

// With the `tracing` feature and a subscriber installed:
// tracing_subscriber::fmt::init();

ticker(Duration::from_secs(1))
    .count()
    .logged("tick", Info)  // emits a tracing event per tick when feature = "tracing"
    .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Cycles(3))
    .unwrap();

```

--------------------------------

### Graph Execution Example with Filtering and Merging

Source: https://docs.rs/wingfoil/4.0.1/index.html

Demonstrates graph execution by creating a stream of numbers, filtering for even and odd numbers, formatting them, and merging the results before printing. The graph runs for a specified duration.

```rust
use wingfoil::*;
use std::time::Duration;

fn main() {
    let period = Duration::from_millis(10);
    let source = ticker(period).count(); // 1, 2, 3 etc
    let is_even = source.map(|i| i % 2 == 0);
    let odds = source
        .filter(is_even.not())
        .map(|i| format!({"i"})).into_bytes());
    let evens = source
        .filter(is_even)
        .map(|i| format!({"i"})).into_bytes());
    merge(vec![odds, evens])
        .print()
        .run(
            RunMode::HistoricalFrom(NanoTime::ZERO),
            RunFor::Duration(period * 5),
        );
}

```

--------------------------------

### Create a new Graph instance

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.Graph.html

Initializes a new Graph with root nodes, a run mode, and a run duration. Ensure root_nodes are properly wrapped in Rc<dyn Node>.

```rust
pub fn new( root_nodes: Vec<Rc<dyn Node>>, run_mode: RunMode, run_for: RunFor, ) -> Graph
```

--------------------------------

### Graph Initialization with Tokio Runtime

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Constructs a new Graph instance with an optional Tokio runtime, allowing for asynchronous operations. It initializes the graph state and then sets up the root nodes.

```APIDOC
## Graph::new_with

### Description
Initializes a new `Graph` with a specified Tokio runtime, root nodes, run mode, and run duration. This is useful for integrating with asynchronous execution environments.

### Method
`Graph::new_with(root_nodes: Vec<Rc<dyn Node>>, tokio_runtime: Arc<tokio::runtime::Runtime>, run_mode: RunMode, run_for: RunFor, start_time: NanoTime) -> Graph`

### Parameters
- `root_nodes` (Vec<Rc<dyn Node>>): A vector of root nodes to be included in the graph.
- `tokio_runtime` (Arc<tokio::runtime::Runtime>): An `Arc`-wrapped Tokio runtime for async operations.
- `run_mode` (RunMode): The execution mode for the graph.
- `run_for` (RunFor): The duration for which the graph should run.
- `start_time` (NanoTime): The specific start time for the graph's execution.

### Returns
A new `Graph` instance configured for asynchronous operations.
```

--------------------------------

### Test: ticked_at_elapsed emits elapsed time

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/graph_state.rs.html

Tests the `ticked_at_elapsed` functionality, ensuring it emits the elapsed time since the start. This is similar to `ticked_at` when the start time is zero.

```rust
#[test]
fn ticked_at_elapsed_emits_elapsed_time() {
    let src: Rc<RefCell<CallBackStream<u64>>> = Rc::new(RefCell::new(CallBackStream::new()));
    src.borrow_mut().push(ValueAt::new(0, NanoTime::new(100)));
    src.borrow_mut().push(ValueAt::new(0, NanoTime::new(250)));

    let elapsed = src.clone().as_node().ticked_at_elapsed().collect();
    elapsed
        .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
        .unwrap();
    // elapsed = time - start_time(0) => same as ticked_at
    let vals: Vec<NanoTime> = elapsed.peek_value().iter().map(|v| v.value).collect();
    assert_eq!(vals, vec![NanoTime::new(100), NanoTime::new(250)]);
}
```

--------------------------------

### Setup AsyncProducerStream Node

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/async_io.rs.html

Configures the `AsyncProducerStream` for execution. It handles different run modes, sets up the sender, and spawns a Tokio task to run the producer future. The task sends messages from the produced stream and handles errors.

```rust
let run_mode = state.run_mode();
        let run_for = state.run_for();
        let mut sender = self
            .sender
            .take()
            .ok_or_else(|| anyhow::anyhow!("sender is already taken"))?;

        match run_mode {
            RunMode::HistoricalFrom(_) => {} // No-op for historical
            RunMode::RealTime => sender.set_notifier(state.ready_notifier()),
        };
        let mut sender = sender.into_async();
        let func = self
            .func
            .take()
            .ok_or_else(|| anyhow::anyhow!("func is already taken"))?;
        let ctx = RunParams {
            run_mode,
            run_for,
            start_time: state.start_time(),
        };
        let fut = async move {
            match func(ctx).await {
                Err(e) => {
                    sender
                        .send_message(Message::Error(std::sync::Arc::new(e)))
                        .await;
                }
                Ok(stream) => {
                    let source = stream.to_message_stream(run_mode).limit(run_mode, run_for);
                    let mut source = Box::pin(source);
                    while let Some(message) = source.next().await {
                        sender.send_message(message).await;
                    }
                }
            }
            sender.close().await;
            Ok(())
        };
        let handle = state.tokio_runtime().spawn(fut);
        self.handle = Some(handle);
        self.receiver_stream.setup(state)?;
        Ok(())
```

--------------------------------

### Graph Initialization

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Constructs a new Graph instance with the specified root nodes, run mode, and run-for duration. It initializes the graph state and then sets up the root nodes.

```APIDOC
## Graph::new

### Description
Initializes a new `Graph` with a given set of root nodes, run mode, and run duration.

### Method
`Graph::new(root_nodes: Vec<Rc<dyn Node>>, run_mode: RunMode, run_for: RunFor) -> Graph`

### Parameters
- `root_nodes` (Vec<Rc<dyn Node>>): A vector of root nodes to be included in the graph.
- `run_mode` (RunMode): The execution mode for the graph.
- `run_for` (RunFor): The duration for which the graph should run.

### Returns
A new `Graph` instance.
```

--------------------------------

### ConstantStream Node Definition

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/constant.rs.html

Defines the ConstantStream struct and its implementation for MutableNode. It holds a value of type T and outputs it. The `cycle` method always returns true, indicating it has completed its work, and the `start` method adds a callback for the graph's start time.

```rust
use crate::types::*;
use derive_new::new;

/// Only ticks once (on the first [Graph](crate::graph::Graph) cycle).
#[derive(new)]
pub(crate) struct ConstantStream<T: Element> {
    value: T,
}

#[node(output = value: T)]
impl<T: Element> MutableNode for ConstantStream<T> {
    fn cycle(&mut self, _state: &mut GraphState) -> anyhow::Result<bool> {
        Ok(true)
    }

    fn start(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
        state.add_callback(state.start_time());
        Ok(())
    }
}
```

--------------------------------

### Initialize Graph with Root Nodes

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Creates a new Graph instance and initializes its state with the provided root nodes, run mode, and run duration.

```rust
pub fn new(root_nodes: Vec<Rc<dyn Node>>, run_mode: RunMode, run_for: RunFor) -> Graph {
        let start_time = run_mode.start_time();
        let state = GraphState::new(run_mode, run_for, start_time);
        let mut graph = Graph { state };
        graph.initialise(root_nodes);
        graph
    }
```

--------------------------------

### GraphState::start_time

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.GraphState.html

Retrieves the start time of the graph execution.

```APIDOC
## GraphState::start_time

### Description
Returns the start time of the graph execution.

### Signature
```rust
pub fn start_time(&self) -> NanoTime
```
```

--------------------------------

### NanoTime::now

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.NanoTime.html

Gets the current time as a NanoTime instance.

```APIDOC
## NanoTime::now

### Description
Gets the current time as a `NanoTime` instance.

### Signature
```rust
pub fn now() -> Self
```
```

--------------------------------

### NodeOperators::ticked_at_elapsed

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/nodes/mod.rs.html

Emits the time of source ticks relative to the start.

```APIDOC
## ticked_at_elapsed

### Description
Returns a stream that emits the time elapsed since the stream's inception (in nanoseconds) for each tick from the source node.

### Signature
```rust
fn ticked_at_elapsed(self: &Rc<Self>) -> Rc<dyn Stream<NanoTime>>
```

### Example
```rust
# use wingfoil::*;
# use std::time::Duration;
// Example output: 0, 1000000000, 2000000000, etc. (depending on tick timing)
ticker(Duration::from_millis(10)).ticked_at_elapsed();
```
```

--------------------------------

### Initialize AsyncChannelSender

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/channel/kanal_chan.rs.html

Creates a new `AsyncChannelSender`. It requires a `kanal::Sender` and an optional `ReadyNotifier`.

```rust
pub fn new(sender: kanal::Sender<Message<T>>, ready_notifier: Option<ReadyNotifier>) -> Self {
    let kanal_sender = Some(sender.to_async());
    Self {
        kanal_sender,
        ready_notifier,
    }
}
```

--------------------------------

### GraphState::elapsed

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.GraphState.html

Retrieves the elapsed time since the graph started.

```APIDOC
## GraphState::elapsed

### Description
Returns the elapsed time since the graph started.

### Signature
```rust
pub fn elapsed(&self) -> NanoTime
```
```

--------------------------------

### Get Run For Duration

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Returns the configured run-for duration of the graph.

```APIDOC
## GraphState::run_for

### Description
Returns the configured duration for which the graph is set to run.

### Method
`run_for(&self) -> RunFor`

### Returns
The `RunFor` duration of the graph.
```

--------------------------------

### GraphState::new

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/struct.GraphState.html

Constructs a new GraphState instance.

```APIDOC
## GraphState::new

### Description
Constructs a new `GraphState` instance.

### Signature
```rust
pub fn new(run_mode: RunMode, run_for: RunFor, start_time: NanoTime) -> Self
```
```

--------------------------------

### Get Run Mode

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Returns the current run mode of the graph.

```APIDOC
## GraphState::run_mode

### Description
Returns the current execution mode of the graph.

### Method
`run_mode(&self) -> RunMode`

### Returns
The `RunMode` of the graph.
```

--------------------------------

### Get Run For Duration

Source: https://docs.rs/wingfoil/4.0.1/src/wingfoil/graph.rs.html

Returns the specified duration for which the graph is intended to run.

```rust
pub fn run_for(&self) -> RunFor {
        self.run_for
    }
```

--------------------------------

### sample

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/trait.Stream.html

Samples its source on each tick of trigger.

```APIDOC
## sample

### Description
Samples its source on each tick of trigger.

### Signature
`fn sample(self: &Rc<Self>, trigger: Rc<dyn Node>) -> Rc<dyn Stream<T>>`

### Parameters
* `trigger`: A node that triggers sampling of the source stream.
```

--------------------------------

### Type ID Implementation

Source: https://docs.rs/wingfoil/4.0.1/wingfoil/enum.Dep.html

Gets the `TypeId` of the current type. This is part of the Any trait implementation.

```rust
fn type_id(&self) -> TypeId
```
Tokens:
Raw