# pipe-io - API Reference
> Authoritative reference for the public API of `pipe-io 1.0.0`.
> Mirrors the rustdoc on docs.rs. Section numbers match `REPS.md`
> section 4. The surface listed here is **frozen**; see
> [Stability](#stability) below.
## Contents
- [Crate root](#crate-root)
- [`pipe_io::source`](#pipe_iosource)
- [`pipe_io::stage`](#pipe_iostage)
- [`pipe_io::sink`](#pipe_iosink)
- [`pipe_io::batch`](#pipe_iobatch)
- [`pipe_io::window` (std)](#pipe_iowindow-std)
- [`pipe_io::error`](#pipe_ioerror)
- [`pipe_io::driver`](#pipe_iodriver)
- [`pipe_io::emit`](#pipe_ioemit)
- [Feature flags](#feature-flags)
- [MSRV](#msrv)
- [Runtime dependencies](#runtime-dependencies)
- [Stability](#stability)
- [Performance](#performance)
## Crate root
| `pipe_io::VERSION` | const | `&'static str`, populated by Cargo at build time. |
| `pipe_io::Pipeline<S>` | struct | A built, runnable pipeline. |
| `pipe_io::PipelineBuilder<T, S, A>` | struct | Typed builder. Carrier type `T` is tracked. |
| `pipe_io::RunStats` | struct | Counters and timing returned by `Pipeline::run`. |
| `pipe_io::Error` | enum | Top-level error. |
| `pipe_io::Result<T>` | alias | `Result<T, Error>`. |
| `pipe_io::StageId` | struct | `&'static str` newtype for stage identification. |
| `pipe_io::StageError` | trait | Marker for error types stages can produce. |
| `pipe_io::StageFailure` | struct | Per-item stage failure (dead-letter material). |
| `pipe_io::BoxError` | alias | `Box<dyn StageError>`. |
| `pipe_io::ErrorPolicy` | enum | Per-stage error policy. |
| `pipe_io::Source` | trait | Re-export from `pipe_io::source`. |
| `pipe_io::Stage` | trait | Re-export from `pipe_io::stage`. |
| `pipe_io::Sink` | trait | Re-export from `pipe_io::sink`. |
| `pipe_io::Emit` | trait | Re-export from `pipe_io::emit`. |
| `pipe_io::EmitError` | enum | Re-export from `pipe_io::emit`. |
| `pipe_io::Driver` | trait | Re-export from `pipe_io::driver`. |
| `pipe_io::Batch<T>` | struct | Owned group of items emitted by `.batch()`. |
| `pipe_io::BatchPolicy` | struct | Batch trigger configuration. |
| `pipe_io::ByteSize` | trait | Opt-in for byte-aware batching. |
| `pipe_io::Window<T>` **(std)** | struct | Owned window emitted by `.window()`. |
| `pipe_io::WindowPolicy` **(std)** | enum | Tumbling / Sliding / Session policies. |
| `pipe_io::Clock` **(std)** | trait | Pluggable wall-clock source. |
| `pipe_io::SystemClock` **(std)** | struct | Default `Clock` backed by `Instant::now`. |
## `pipe_io::source`
```rust
pub trait Source {
type Item;
type Error: StageError;
fn pull(&mut self) -> Result<Option<Self::Item>, Self::Error>;
fn close(&mut self) -> Result<(), Self::Error> { Ok(()) }
}
```
Adapters:
| `IterSource<I: Iterator>` | struct | Wraps any `IntoIterator`. `Error = Infallible`. |
| `FnSource<F, T, E>` | struct | Wraps `FnMut() -> Result<Option<T>, E>`. |
| `ChannelSource<T>` **(std)** | struct | Wraps `mpsc::Receiver<T>`. |
| `ReaderSource<R: io::Read>` **(std)** | struct | Line-buffered. `Error = std::io::Error`. |
| `Infallible` | enum | Uninhabited error type for sources that cannot fail. |
## `pipe_io::stage`
```rust
pub trait Stage {
type Input;
type Output;
type Error: StageError;
fn process(
&mut self,
item: Self::Input,
out: &mut dyn Emit<Item = Self::Output>,
) -> Result<(), Self::Error>;
fn flush(
&mut self,
_out: &mut dyn Emit<Item = Self::Output>,
) -> Result<(), Self::Error> { Ok(()) }
}
```
Stages emit zero or more outputs per input via `Emit`. Closure-based
adapters (`map`, `filter`, `filter_map`, `flat_map`, `inspect`,
`try_map`) live on `PipelineBuilder`.
## `pipe_io::sink`
```rust
pub trait Sink {
type Item;
type Error: StageError;
fn write(&mut self, item: Self::Item) -> Result<(), Self::Error>;
fn flush(&mut self) -> Result<(), Self::Error> { Ok(()) }
fn close(&mut self) -> Result<(), Self::Error> { Ok(()) }
}
```
Adapters:
| `NullSink<T>` | struct | Discards every item. |
| `FnSink<F, T, E>` | struct | Wraps `FnMut(T) -> Result<(), E>`. |
| `VecSink<T>` **(std)** | struct | Collects into a `Vec<T>`. Items accessible via `.handle()` (cloneable, `Send + Sync`). |
| `SharedHandle<T>` **(std)** | struct | `Arc<Mutex<Vec<T>>>` wrapper returned by `VecSink::handle`. Methods: `take`, `len`, `is_empty`, `clone`. |
| `ChannelSink<T>` **(std)** | struct | Wraps `mpsc::SyncSender<T>`. |
| `ChannelSinkError` **(std)** | enum | `Disconnected` variant. |
| `WriterSink<W: io::Write>` **(std)** | struct | Line-writes `String` items into any writer. |
## `pipe_io::batch`
```rust
pub struct BatchPolicy { /* opaque */ }
impl BatchPolicy {
pub const fn new() -> Self;
pub const fn max_items(self, n: usize) -> Self;
pub const fn max_bytes(self, n: usize) -> Self;
pub const fn max_age(self, age: Duration) -> Self; // std only
pub const fn items_limit(&self) -> Option<usize>;
pub const fn bytes_limit(&self) -> Option<usize>;
pub const fn age_limit(&self) -> Option<Duration>; // std only
pub const fn has_trigger(&self) -> bool;
}
pub trait ByteSize { fn byte_size(&self) -> usize; }
pub struct Batch<T> {
pub fn new(items: Vec<T>) -> Self;
pub fn len(&self) -> usize;
pub fn is_empty(&self) -> bool;
pub fn iter(&self) -> slice::Iter<'_, T>;
pub fn into_inner(self) -> Vec<T>;
}
// Deref<Target = [T]>, IntoIterator for Batch<T> and &Batch<T>.
```
Blanket `ByteSize` impls: `&str`, `String`, `Vec<u8>`, `&[u8]`.
Insert a batching stage via `PipelineBuilder::batch(policy)` (count
and age triggers) or `PipelineBuilder::batch_bytes(policy)` (when the
policy has a `max_bytes` trigger; requires `T: ByteSize`).
## `pipe_io::window` (std)
```rust
pub trait Clock: Send {
fn now(&self) -> Instant;
}
#[derive(Default, Clone, Copy)]
pub struct SystemClock;
impl Clock for SystemClock { /* wraps Instant::now */ }
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum WindowPolicy {
Tumbling { size: Duration },
Sliding { size: Duration, slide: Duration },
Session { idle: Duration },
}
pub struct Window<T> {
pub fn new(items: Vec<T>, start: Instant, end: Instant) -> Self;
pub fn items(&self) -> &[T];
pub fn len(&self) -> usize;
pub fn is_empty(&self) -> bool;
pub fn start(&self) -> Instant;
pub fn end(&self) -> Instant;
pub fn into_inner(self) -> Vec<T>;
}
// IntoIterator for Window<T> and &Window<T>.
```
Install via `PipelineBuilder::window(policy)` (default
`SystemClock`) or `PipelineBuilder::window_with(policy, clock)` for
a user-supplied clock. Both require `T: Clone` because the sliding
policy duplicates items across overlapping windows. Windows close on
the next item arriving after the close condition fires, or at
end-of-stream.
## `pipe_io::error`
```rust
pub trait StageError: Debug + Display + Send + Sync + 'static {}
impl<T> StageError for T where T: Debug + Display + Send + Sync + 'static {}
pub type BoxError = Box<dyn StageError>;
pub type Result<T> = core::result::Result<T, Error>;
#[non_exhaustive]
pub enum Error {
Source { stage: StageId, source: BoxError },
Stage { stage: StageId, source: BoxError },
Sink { stage: StageId, source: BoxError },
Buffer { stage: StageId, kind: BufferErrorKind },
Cancelled,
Closed,
}
impl Error {
pub fn stage(&self) -> Option<StageId>;
}
pub enum BufferErrorKind { Full, Closed }
#[derive(Default)]
pub enum ErrorPolicy {
#[default] FailFast,
Continue,
DeadLetter, // std: routes via PipelineBuilder::dead_letter; degrades to Continue if no sink installed
}
#[non_exhaustive]
pub struct StageFailure {
pub stage: StageId,
pub source: BoxError,
}
impl StageFailure {
pub fn new(stage: StageId, source: BoxError) -> Self;
}
```
`StageError`'s blanket impl admits every `std::error::Error` type
plus any other `Debug + Display + Send + Sync + 'static` type. The
`core::error::Error` super-trait was avoided because it stabilized
in Rust 1.81 and the crate's MSRV is 1.75.
## `pipe_io::driver`
```rust
#[derive(Default, Clone, Copy)]
pub struct RunStats {
pub items_in: u64,
pub duration: Duration, // std only
}
pub trait Driver {
fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
where
S: Source + Send + 'static,
S::Item: Send + 'static,
S::Error: Send + 'static;
}
#[derive(Default, Clone, Copy)]
pub struct SyncDriver;
impl SyncDriver {
pub const fn new() -> Self;
// Inherent method with looser bound (no Send required); use this
// when driving a non-Send source on the current thread.
pub fn run<S: Source>(self, pipeline: Pipeline<S>) -> Result<RunStats>;
}
impl Driver for SyncDriver { /* delegates to inherent run */ }
#[derive(Default, Clone, Copy)]
pub struct ThreadedDriver; // std only
impl ThreadedDriver {
pub const fn new() -> Self;
pub fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
where
S: Source + Send + 'static,
S::Item: Send + 'static,
S::Error: Send + 'static;
}
impl Driver for ThreadedDriver { /* delegates to inherent run */ }
```
Consumers select a driver by calling `Pipeline::run` (sync),
`Pipeline::run_threaded` (std), or `Pipeline::run_with(driver)`
for any `Driver` impl. The trait is not sealed; external
executors (tokio, rayon, custom thread farm) can implement it.
## `pipe_io::emit`
```rust
pub trait Emit {
type Item;
fn emit(&mut self, item: Self::Item) -> Result<(), EmitError>;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum EmitError {
Closed,
WouldBlock,
}
```
## Builder surface
```rust
impl<S: Source> Pipeline<S> {
pub fn from_source(source: S) -> PipelineBuilder<S::Item, S, _>;
pub fn from_iter<II: IntoIterator>(iter: II) -> PipelineBuilder<II::Item, IterSource<II::IntoIter>, _>;
pub fn run(self) -> Result<RunStats>;
pub fn run_threaded(self) -> Result<RunStats>; // std only
pub fn run_with<D: Driver>(self, driver: D) -> Result<RunStats>
where S: Send, S::Item: Send, S::Error: Send;
}
impl<T, S, Acc> PipelineBuilder<T, S, Acc>
where
S: Source + 'static,
T: Send + 'static,
S::Item: Send + 'static,
{
pub fn stage_id<I: Into<StageId>>(self, id: I) -> Self;
pub fn on_error(self, policy: ErrorPolicy) -> Self;
pub fn map<U, F>(self, f: F) -> PipelineBuilder<U, S, _>
where F: FnMut(T) -> U + Send + 'static;
pub fn filter<F>(self, pred: F) -> PipelineBuilder<T, S, _>
where F: FnMut(&T) -> bool + Send + 'static;
pub fn filter_map<U, F>(self, f: F) -> PipelineBuilder<U, S, _>
where F: FnMut(T) -> Option<U> + Send + 'static;
pub fn flat_map<U, F, II>(self, f: F) -> PipelineBuilder<U, S, _>
where II: IntoIterator<Item = U>, F: FnMut(T) -> II + Send + 'static;
pub fn inspect<F>(self, f: F) -> PipelineBuilder<T, S, _>
where F: FnMut(&T) + Send + 'static;
pub fn try_map<U, F, E>(self, f: F) -> PipelineBuilder<U, S, _>
where E: StageError, F: FnMut(T) -> Result<U, E> + Send + 'static;
pub fn stage<St: Stage<Input = T>>(self, stage: St) -> PipelineBuilder<St::Output, S, _>;
pub fn batch(self, policy: BatchPolicy) -> PipelineBuilder<Batch<T>, S, _>;
pub fn batch_bytes(self, policy: BatchPolicy) -> PipelineBuilder<Batch<T>, S, _>
where T: ByteSize;
// std only
pub fn window(self, policy: WindowPolicy) -> PipelineBuilder<Window<T>, S, _>
where T: Clone;
pub fn window_with<C: Clock>(self, policy: WindowPolicy, clock: C)
-> PipelineBuilder<Window<T>, S, _>
where T: Clone;
// std only - installs a dead-letter sink for ErrorPolicy::DeadLetter routing.
// Can be called before or after the failing stages; replaces any previously
// installed sink. If `run` is called without one installed, DeadLetter
// failures silently drop (same as Continue).
pub fn dead_letter<Sk: Sink<Item = StageFailure>>(self, sink: Sk) -> Self;
pub fn sink<Sk: Sink<Item = T>>(self, sink: Sk) -> Pipeline<S>;
}
```
The `Acc` parameter is an anonymous closure type the builder
threads through; the user never names it.
## Feature flags
| `std` | yes | `ThreadedDriver`, `Pipeline::run_threaded`, `ChannelSource`, `ReaderSource`, `ChannelSink`, `WriterSink`, `VecSink`, `Window` / `Clock` / `SystemClock`, `BatchPolicy::max_age`, dead-letter routing. |
## MSRV
`1.75`. Locked from `1.0.0` onward; a bump requires a minor
version increment and a CHANGELOG entry under `### Changed`.
## Runtime dependencies
Zero. Built on `core` + `alloc` (always) plus `std` when the
`std` feature is enabled. `proptest` is a dev-only dependency.
## Stability
From `1.0.0` forward, the public surface above is frozen.
Patch releases (`1.0.x`) ship bug fixes and doc improvements
only. Minor releases (`1.x.0`) add to the surface but never
remove or rename. `cargo-semver-checks` is a hard CI gate.
The `Error` and `StageFailure` types are `#[non_exhaustive]`;
match arms on them must include a wildcard pattern. See
[`REPS.md`](../REPS.md) section 8 for the binding policy.
## Performance
See [`BENCH.md`](BENCH.md) for benchmark methodology and measured
numbers. Headline (Windows, x86_64, release with `lto = "thin"`):
| source -> null sink | ~500 |
| source -> map -> sink | ~260 |
| source -> map -> filter -> map -> sink | ~170 |
| source -> batch(100) -> sink | ~140 |
| threaded driver: source -> map -> sink | ~21 |
Each stage adds roughly 1-2 ns per item (one vtable dispatch through
the boxed stage chain). The threaded driver pays for thread spawn
plus closure transfer; it amortizes for long-running pipelines or
when source pull blocks.
## Quick example
```rust
use pipe_io::{Pipeline, sink::VecSink, BatchPolicy};
let sink = VecSink::<Vec<i64>>::new();
let handle = sink.handle();
Pipeline::from_iter(1..=10)
.map(|n: i32| i64::from(n) * 10)
.filter(|n: &i64| *n > 30)
.batch(BatchPolicy::new().max_items(2))
.map(|b: pipe_io::Batch<i64>| b.into_inner())
.sink(sink)
.run()
.unwrap();
assert_eq!(handle.take(), vec![vec![40, 50], vec![60, 70], vec![80, 90], vec![100]]);
```