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

//! The [`Stage`] trait.
//!
//! Stages receive items one at a time and emit zero or more outputs via
//! an [`Emit`] handle. The closure-based adapters (`map`, `filter`, ...)
//! are exposed as methods on [`crate::PipelineBuilder`].

use crate::emit::Emit;
use crate::error::StageError;

/// Transform stage in a pipeline.
///
/// Stages emit zero or more outputs per input via the `out` handle.
/// 1:1 maps emit once per input, filters emit 0..1 per input,
/// batching/windowing emit 0+ per input and additionally use
/// [`Stage::flush`] to emit a final partial group at end-of-stream.
pub trait Stage {
    /// Type of item this stage consumes.
    type Input;
    /// Type of item this stage emits.
    type Output;
    /// Error type the stage can produce.
    type Error: StageError;

    /// Process a single input item.
    ///
    /// # Errors
    ///
    /// Returns `Err(Self::Error)` on stage failure. The driver wraps
    /// this in [`crate::Error::Stage`].
    fn process(
        &mut self,
        item: Self::Input,
        out: &mut dyn Emit<Item = Self::Output>,
    ) -> Result<(), Self::Error>;

    /// Emit any buffered output at end-of-stream.
    ///
    /// Default impl does nothing.
    ///
    /// # Errors
    ///
    /// Returns `Err(Self::Error)` on stage failure.
    fn flush(&mut self, _out: &mut dyn Emit<Item = Self::Output>) -> Result<(), Self::Error> {
        Ok(())
    }
}