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

//! Implementing the [`Driver`] trait for a custom executor.
//!
//! `LoggingDriver` wraps `SyncDriver` and records the wall-clock
//! duration plus the items processed. Real custom drivers can plug
//! into tokio, rayon, glommio, or any other scheduling primitive.
//!
//! Run with:
//!
//! ```text
//! cargo run --example custom_driver
//! ```

use std::time::Instant;

use pipe_io::driver::{Driver, RunStats, SyncDriver};
use pipe_io::{sink::VecSink, Pipeline, Result};

struct LoggingDriver {
    name: &'static str,
}

impl Driver for LoggingDriver {
    fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
    where
        S: pipe_io::Source + Send + 'static,
        S::Item: Send + 'static,
        S::Error: Send + 'static,
    {
        let start = Instant::now();
        let result = SyncDriver::new().run(pipeline);
        match &result {
            Ok(stats) => {
                println!(
                    "[{}] ran {} items in {:?}",
                    self.name,
                    stats.items_in,
                    start.elapsed()
                );
            }
            Err(e) => {
                println!("[{}] failed after {:?}: {e}", self.name, start.elapsed());
            }
        }
        result
    }
}

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

    Pipeline::from_iter(0..1_000)
        .map(|n: i32| n * 2)
        .filter(|n: &i32| *n % 3 == 0)
        .sink(sink)
        .run_with(LoggingDriver { name: "demo" })
        .expect("pipeline run");

    let out = handle.take();
    println!("collected {} items", out.len());
}