rskit-pipeline 0.1.0-alpha.2

Composable async data pipelines via futures::Stream extension operators
Documentation

rskit-pipeline — Async Stream Operators

RskitStreamExt adds lazy, backpressure-aware async operators to any futures::Stream.

CI crates.io docs.rs License: MIT MSRV: 1.91

Features

Operators available on any futures::Stream via RskitStreamExt:

  • rmap — async map
  • rflatmap — async flat-map
  • rfilter — predicate filter
  • rtap — side-effect tap
  • rreduce — async fold/reduce
  • rparallel — concurrent execution with bounded parallelism
  • rfan_out — broadcast to multiple async functions
  • rmerge — merge two streams
  • rpartition — split into matching and non-matching streams
  • rbatch — collect into fixed-size/time-limited batches
  • rdebounce — debounce with a quiet period
  • rthrottle — rate-limited throughput
  • rtumbling_window — non-overlapping time windows
  • rsliding_window — overlapping item-count windows
  • rdistinct — first occurrence only
  • rtake / rskip — bounded prefix operators

Operator reference

Operator rskit API Semantics
map RskitStreamExt::rmap Transform each item with a fallible async function.
filter RskitStreamExt::rfilter Keep items matching a synchronous predicate.
batch RskitStreamExt::rbatch Emit up to size items, or the partial batch when timeout elapses/upstream ends.
window RskitStreamExt::rtumbling_window Emit non-overlapping windows bounded by duration and item count.
sliding RskitStreamExt::rsliding_window Emit overlapping item-count windows advancing by step.
fan_out RskitStreamExt::rfan_out Apply a bounded number of async functions to each item, poll branch futures concurrently in the stream task, and short-circuit on the first branch error.
parallel RskitStreamExt::rparallel Process items concurrently with bounded parallelism; output order is not guaranteed.
merge RskitStreamExt::rmerge, merge Yield items from whichever input stream is ready first.
partition RskitStreamExt::rpartition Route each item to matching or remainder stream; one closed side does not close the other.
throttle RskitStreamExt::rthrottle Emit at most one item per interval.
debounce RskitStreamExt::rdebounce Wait for a quiet period, then emit the latest item.
distinct RskitStreamExt::rdistinct Emit the first occurrence of each item.
take RskitStreamExt::rtake Emit only the first n items.
skip RskitStreamExt::rskip Drop the first n items.

Sources

Source rskit API Semantics
from_slice from_slice Stream a Vec<T> in order.
from_fn from_fn Lazily generate items from an async closure until it returns None.
from_channel from_channel Wrap a tokio::sync::mpsc::Receiver as a stream.
broadcaster Broadcaster<T> Bounded, cancellable one-to-many fan-out: each subscribe returns its own bounded BroadcastStream; lagging subscribers drop overflow (never block the broadcaster), disconnected ones are pruned lazily. Cancellation via CancellationToken.

Broadcaster<T> is the canonical owner for "observe a backend → bounded, cancellable stream of typed change events" — config reloads, service discovery, cache invalidation, secret rotation — instead of each crate reinventing a per-subscriber fan-out.

Usage

[dependencies]
rskit-pipeline = "0.1.0-alpha.1"
rskit-errors = "0.1.0-alpha.1"

use futures_util::StreamExt;
use rskit_pipeline::{from_slice, RskitStreamExt};

let results = from_slice(vec![1u32, 2, 3, 4, 5])
    .rfilter(|&n| n % 2 == 0)
    .rmap(|n| async move { Ok::<_, rskit_errors::AppError>(n * 10) })
    .collect::<Vec<_>>()
    .await;
// [Ok(20), Ok(40)]

See Also

Main repository README