pipe-io 1.0.0

Typed source-transform-sink pipelines with backpressure, batching, windowing, and per-stage error isolation. A lightweight runtime-agnostic stream processor for in-process workloads. The missing middle ground between raw iterators and full distributed stream processing.
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

//! Driving a pipeline on a background thread via `ThreadedDriver`.
//!
//! The calling thread blocks on `run_threaded()` until the worker
//! finishes. Use this when you want the pipeline body to run off the
//! current thread (for example, alongside an event loop that owns
//! the main thread) without bringing in an async runtime.
//!
//! Run with:
//!
//! ```text
//! cargo run --example threaded
//! ```

use pipe_io::{sink::VecSink, Pipeline};

fn main() {
    let sink = VecSink::<i64>::new();
    let handle = sink.handle();

    let stats = Pipeline::from_iter(1i32..=5)
        .map(|n: i32| i64::from(n) * 100)
        .filter(|n: &i64| *n >= 200)
        .sink(sink)
        .run_threaded()
        .expect("pipeline run on worker thread");

    println!("processed {} items in {:?}", stats.items_in, stats.duration);
    println!("output: {:?}", handle.take());
}