Operator Reference

Complete index of Datum's stream operators, grouped by category and cross-referenced to their Akka Streams equivalents. Status follows the M2 ledger: ✅ shipped · 🟡 partial · ⏭️ deferred · ❌ won't implement.

For narrative context on the major areas, see the corresponding guides: Graphs · Buffers & Rate · Error Handling · Futures Interop · Pipelining And Parallelism · Context Propagation · Actors Interop · Custom Stream Processing · Dynamic Streams · Streaming IO · Testing Streams · Substreams

Source operators

Akka operator	Datum operator	Notes
`Source.empty`	`Source::empty()`	✅
`Source.single`	`Source::single(item)`	✅
`Source.repeat`	`Source::repeat(item)`	✅
`Source.failed`	`Source::failed(error)`	✅
`Source.fromIterator`	`Source::from_fn_iter(factory)`	✅
`Source.from` / `fromIterable`	`Source::from_iterable(items)`	✅
`Source.range`	`Source::from_iterable(a..b)`	✅ `range(a, b)` = `from_iterable(a..b)`
`Source.future`	`Source::future(fut_factory)`	✅
`Source.futureSource`	`Source::future_source(fut_factory)`	✅
`Source.maybe`	`Source::maybe()`	✅ returns `(MaybeHandle<T>, Source<T>)`
`Source.never`	`Source::never()`	✅
`Source.tick`	`Source::tick(initial, interval, element)`	✅
`Source.cycle`	`Source::cycle(factory)`	✅ factory called each cycle
`Source.unfold`	`Source::unfold(initial, f)`	✅
`Source.unfoldAsync`	`Source::unfold_async(initial, f)`	✅ Tokio-dispatched
`Source.unfoldResource`	`Source::unfold_resource(create, read, close)`	✅
`Source.unfoldResourceAsync`	`Source::unfold_resource_async(create, read, close)`	✅ Tokio-dispatched
`Source.lazily` / `lazySource`	`Source::lazy_source(factory)`	✅ defers construction to materialization
`Source.lazySingle`	`Source::lazy_single(factory)`	✅
`Source.lazyFuture`	`Source::lazy_future(factory)`	✅
`Source.lazyFutureSource`	`Source::lazy_future_source(factory)`	✅
`Source.combine`	`Source::combine(sources, strategy)`	✅ `SourceCombineStrategy`
`Source.zipN`	`Source::zip_n(sources)`	✅
`Source.zipWithN`	`Source::zip_with_n(sources, zipper)`	✅
`Source.queue`	`Source::queue(capacity, strategy)` / `Source::queue_bounded(capacity)`	✅ `SourceQueue<T>` / `BoundedSourceQueue<T>` are the materialized values
`Source.asSourceWithContext`	`Source::as_source_with_context(extract)`	✅
`Source.mergePrioritizedN`	`Source::merge_prioritized_n(sources_with_priority)`	✅
`Source.asSubscriber` / `fromPublisher`	—	❌ Reactive Streams — out of scope; use Tokio futures
`Source.fromCompletionStage` / JVM `fromJavaStream`	—	❌ JVM-only; use `Source::future` / `Source::from_iterable`

Sink operators

Akka operator	Datum operator	Notes
`Sink.ignore`	`Sink::ignore()`	✅
`Sink.head`	`Sink::head()`	✅ fails on empty stream
`Sink.headOption`	`Sink::head_option()`	✅
`Sink.last`	`Sink::last()`	✅
`Sink.lastOption`	`Sink::last_option()`	✅
`Sink.seq` / `collect`	`Sink::collect()`	✅ returns `Vec<T>`
`Sink.collection`	`Sink::collection()`	✅
`Sink.takeLast`	`Sink::take_last(n)`	✅
`Sink.fold`	`Sink::fold(zero, f)`	✅
`Sink.reduce`	`Sink::reduce(f)`	✅
`Sink.foreach`	`Sink::foreach(f)`	✅
`Sink.foreachAsync`	`Sink::foreach_async(parallelism, f)`	✅
`Sink.cancelled`	`Sink::cancelled()`	✅
`Sink.never`	`Sink::never()`	✅
`Sink.onComplete`	`Sink::on_complete(callback)`	✅
`Sink.futureSink`	`Sink::future_sink(factory)`	✅
`Sink.lazySink`	`Sink::lazy_sink(factory)`	✅
`Sink.lazyFutureSink`	`Sink::lazy_future_sink(factory)`	✅
`Sink.fromMaterializer` / `setup`	`Sink::from_materializer(f)` / `Sink::setup(f)`	✅
`Sink.preMaterialize`	`Sink::pre_materialize()`	✅
`Sink.combine`	`Sink::combine(sinks, strategy)`	✅ `SinkCombineStrategy`
`Sink.queue`	`Sink::queue()`	✅ `SinkQueue`
`Sink.asPublisher` / `fromSubscriber`	—	❌ Reactive Streams — out of scope
`Sink.asInputStream` / `asOutputStream`	`StreamConverters::as_input_stream(read_timeout)` / `StreamConverters::as_output_stream(write_timeout)`	✅ materializes `InputStreamHandle` (`impl Read`) / `OutputStreamHandle` (`impl Write`); blocking IO bridge
JVM `javaCollector*` / `asJavaStream`	—	❌ JVM-only; use `Sink::collect()`

Flow operators (simple / transforming)

Akka operator	Datum operator	Notes
`map`	`.map(f)`	✅
`filter`	`.filter(predicate)`	✅
`filterNot`	`.filter_not(predicate)`	✅
`collect`	`.filter_map(f)`	✅ Rust-native rename; `collect` is a keyword conflict
`mapConcat`	`.map_concat(f)`	✅
`grouped`	`.grouped(size)`	✅
`sliding`	`.sliding(size, step)`	✅
`scan`	`.scan(seed, f)`	✅
`fold`	`.fold(zero, f)`	✅
`reduce`	`.reduce(f)`	✅
`take`	`.take(n)`	✅
`takeWhile`	`.take_while(predicate)`	✅
`drop`	`.drop(n)`	✅
`dropWhile`	`.drop_while(predicate)`	✅
`limit`	`.limit(max)`	✅
`statefulMap`	`.stateful_map(seed, f)`	✅
`statefulMapConcat`	`.stateful_map_concat(seed, f)`	✅
`intersperse`	`.intersperse(inject)`	✅
`flattenOptional`	`.flatten_optional()`	✅
`groupedWeighted`	`.grouped_weighted(max_weight, cost_fn)`	✅
`limitWeighted`	`.limit_weighted(max_weight, cost_fn)`	✅
`contramap`	`.contramap(f)`	✅
`log` / `logWithMarker`	`.monitor(callback)`	✅ logging via closure callback
`monitor`	`.monitor(callback)`	✅
`watchTermination`	`.watch_termination(materialize_callback)`	✅
`foldAsync`	`.fold_async(zero, f)`	✅ Tokio-dispatched
`scanAsync`	`.scan_async(seed, f)`	✅ Tokio-dispatched
`mapWithResource`	`.map_with_resource(create, f, close)`	✅
`futureFlow`	`Flow::future_flow(factory)`	✅
`lazyFlow`	`Flow::lazy_flow(factory)`	✅
`lazyFutureFlow`	`Flow::lazy_future_flow(factory)`	✅
`.async` boundary	`.async_boundary()` / `.r#async()`	✅ bounded Ractor-backed handoff between linear fused regions
`throttle`	`.throttle(elements, per, burst, mode)`	✅ `ThrottleMode::Shaping` / `Enforcing`
`detach`	`.detach()`	✅ decouples fused regions
`asFlowWithContext`	`.as_flow_with_context(collapse, extract)`	✅
`fromSinkAndSource`	`Flow::from_sink_and_source(sink, source)`	✅
`fromSinkAndSourceCoupled`	`Flow::from_sink_and_source_coupled(sink, source)`	✅
`collectType`	—	✅ equiv: use `.filter_map(

Async operators

Akka operator	Datum operator	Notes
`mapAsync`	`.map_async(parallelism, f)`	✅ ordered, Tokio-dispatched
`mapAsyncUnordered`	`.map_async_unordered(parallelism, f)`	✅ unordered, Tokio-dispatched
`mapAsyncPartitioned`	`.map_async_partitioned(parallelism, per_partition, partition_fn, f)`	✅ per-partition ordering

Time-aware operators

Akka operator	Datum operator	Notes
`delay`	`.delay(duration, strategy)`	✅ `DelayOverflowStrategy`
`delayWith`	`.delay_with(supplier, strategy)`	✅ per-element delay
`initialDelay`	`.initial_delay(duration)`	✅
`dropWithin`	`.drop_within(timeout)`	✅
`takeWithin`	`.take_within(timeout)`	✅
`groupedWithin`	`.grouped_within(max_number, interval)`	✅
`groupedWeightedWithin`	`.grouped_weighted_within(max_weight, interval, cost_fn)`	✅
`idleTimeout`	`.idle_timeout(timeout)`	✅
`backpressureTimeout`	`.backpressure_timeout(timeout)`	✅
`completionTimeout`	`.completion_timeout(timeout)`	✅
`initialTimeout`	`.initial_timeout(timeout)`	✅
`keepAlive`	`.keep_alive(timeout, inject)`	✅

Backpressure & rate operators

Akka operator	Datum operator	Notes
`buffer`	`.buffer(size, strategy)`	✅ all `OverflowStrategy` variants
`conflate`	`.conflate(aggregate)`	✅
`conflateWithSeed`	`.conflate_with_seed(seed, aggregate)`	✅
`expand`	`.expand(expand)`	✅
`extrapolate`	`.extrapolate(expand, initial)`	✅
`batch`	`.batch(max, seed, aggregate)`	✅
`batchWeighted`	`.batch_weighted(max, cost_fn, seed, aggregate)`	✅
`aggregateWithBoundary`	`.aggregate_with_boundary(allocate, aggregate, harvest, boundary)`	✅

OverflowStrategy variants: DropHead, DropTail, DropBuffer, DropNew, Backpressure, Fail.

Substream operators

Akka operator	Datum operator	Notes
`flatMapConcat`	`.flat_map_concat(f)`	✅
`flatMapMerge`	`.flat_map_merge(breadth, f)`	✅
`flatMapPrefix`	`.flat_map_prefix(n, f)`	✅
`groupBy`	`.group_by(max_substreams, key_fn, eager_complete)`	✅ closed keys are retained until parent completes
`prefixAndTail`	`.prefix_and_tail(n)`	✅ emits `(Vec<T>, Source<T>)`
`splitWhen`	`.split_when(predicate)`	✅ emits `Source<T>` per segment
`splitAfter`	`.split_after(predicate)`	✅ emits `Source<T>` per segment

See Substreams guide for the SubFlow-vs-nested-source behavioral difference.

Fan-in operators

Akka operator	Datum operator	Notes
`merge`	`Merge<T>::new(n)` (graph junction)	✅
`mergePreferred`	`MergePreferred<T>::new(n_secondary)`	✅
`mergePrioritized`	`MergePrioritized<T>::new(weights)`	✅
`mergePrioritizedN`	`Source::merge_prioritized_n(sources)`	✅
`zip`	`Zip<L, R>::new()` (graph junction)	✅
`concat`	`.concat(source)` / `Concat<T>::new(n)`	✅
`concatLazy`	`.concat_lazy(source)`	✅
`concatAllLazy`	`.concat_all_lazy(sources)`	✅
`prepend`	`.prepend(source)`	✅
`prependLazy`	`.prepend_lazy(source)`	✅
`orElse`	`.or_else(secondary)`	✅
`interleave`	`.interleave(source, segment_size)` / `Interleave<T>::new(n, segment_size)`	✅
`interleaveAll`	`.interleave_all(sources, segment_size, eager_close)`	✅
`mergeSorted`	`.merge_sorted(source)` / `MergeSorted<T>::new()`	✅
`MergeSequence`	`MergeSequence<T>::new(n, extract_seq)`	✅
`mergeLatest`	`.merge_latest(source, eager_complete)` / `MergeLatest<T>::new(inputs, eager_complete)`	✅
`mergeAll`	`.merge_all(sources, eager_complete)`	✅
`zipWith`	`.zip_with(source, f)`	✅
`zipLatest`	`.zip_latest(source)`	✅
`zipLatestWith`	`.zip_latest_with(source, f)`	✅
`zipWithIndex`	`.zip_with_index()`	✅ emits `(T, u64)`
`zipAll`	`.zip_all(source, this_default, that_default)`	✅
`zipN`	`Source::zip_n(sources)`	✅
`zipWithN`	`Source::zip_with_n(sources, zipper)`	✅

Fan-out operators

Akka operator	Datum operator	Notes
`Broadcast`	`Broadcast<T>::new(n)`	✅
`Balance`	`Balance<T>::new(n)`	✅
`Partition`	`Partition<T>::new(n, f)`	✅
`Unzip`	`Unzip<A, B>::new()`	✅
`UnzipWith`	`UnzipWith<In, A, B>::new(f)`	✅
`alsoTo`	`.also_to(sink)`	✅
`alsoToAll`	`.also_to_all(sinks)`	✅
`divertTo`	`.divert_to(sink, predicate)`	✅
`wireTap`	`.wire_tap(sink)`	✅

Actor interop operators

See Actors Interop guide for usage examples.

Akka operator	Datum operator	Notes
`ActorFlow.ask`	`ActorFlow::ask(actor_ref, parallelism, timeout, make_msg)`	✅
`ActorFlow.askWithStatus`	`ActorFlow::ask_with_status(actor_ref, parallelism, timeout, make_msg)`	✅ unwraps `ActorStatus::Ok`/`Err`
`ActorFlow.askWithContext`	`ActorFlow::ask_with_context(actor_ref, parallelism, timeout, make_msg)`	✅ preserves context
`ActorFlow.askWithStatusAndContext`	`ActorFlow::ask_with_status_and_context(actor_ref, parallelism, timeout, make_msg)`	✅
`ActorSource` (`actorRef`)	`ActorSource::actor_ref()`	✅ typed `ActorSourceMessage<T>` protocol
`ActorSource` (`actorRef` backpressured)	`ActorSource::actor_ref_with_backpressure()`	✅
`ActorSource` typed	`ActorSource::typed()`	✅
`ActorSink` (`actorRef`)	`ActorSink::actor_ref(...)`	✅
`ActorSink` (`actorRef` backpressured)	`ActorSink::actor_ref_with_backpressure(...)`	✅
`ActorSink` typed	`ActorSink::typed()` / `ActorSink::typed_with_backpressure()`	✅
PubSub `source` / `sink`	`ActorPubSub::source(group)` / `ActorPubSub::sink(group)`	✅
`watch`	`ActorFlow::watch(actor_ref)`	✅ emits pass-through elements; terminates on actor exit
`StreamRefs.sourceRef`	`StreamRefs::source_ref()`	✅ materializes `SourceRef<T>`
`StreamRefs.sinkRef`	`StreamRefs::sink_ref()`	✅ materializes `SinkRef<T>`

Error-handling operators

See Error Handling guide.

Akka operator	Datum operator	Notes
`recover`	`.recover(f)`	✅
`recoverWith`	`.recover_with(f)`	✅
`recoverWithRetries`	`.recover_with_retries(retries, f)`	✅
`mapError`	`.map_error(f)`	✅
`onErrorComplete`	`.on_error_complete()`	✅
Supervision deciders	`Supervision::stopping_decider()` / `resuming_decider()` / `restarting_decider()`	✅
Fallible callbacks	`.try_map(f)` (and equivalents for `filter`, `filter_map`, `map_concat`, `fold`, `reduce`, `scan`, `stateful_map`, `stateful_map_concat`)	✅ unsupervised `_result` names remain deprecated aliases
Supervised callbacks	`.map_result_with_supervision(f, decider)` (and equivalents for `filter_map`, `fold`, `scan`, `reduce`, `foreach`)	✅ supervision stays explicit
`RestartSource.withBackoff`	`RestartSource::with_backoff(settings, factory)`	✅
`RestartSource.onFailuresWithBackoff`	`RestartSource::on_failures_with_backoff(settings, factory)`	✅
`RestartFlow.withBackoff`	`RestartFlow::with_backoff(settings, factory)`	✅
`RestartFlow.onFailuresWithBackoff`	`RestartFlow::on_failures_with_backoff(settings, factory)`	✅
`RestartSink.withBackoff`	`RestartSink::with_backoff(settings, factory)`	✅ wraps sinks that materialize `StreamCompletion<NotUsed>`
`RetryFlow.withBackoff`	`RetryFlow::with_backoff(min, max, factor, f)`	✅
`RetryFlow.withBackoffAndContext`	`RetryFlow::with_backoff_and_context(...)`	✅

IO & compression operators

See Streaming IO guide.

Akka operator	Datum operator	Notes
`gzip` / `gunzip`	`Compression::gzip()` / `Compression::gunzip()`	✅ via `flate2`
`deflate` / `inflate`	`Compression::deflate()` / `Compression::inflate()`	✅
`fromPath` / `toPath`	`FileIO::from_path(path)` / `FileIO::to_path(path)`	✅ sync `std::fs`
Tokio async file IO	`TokioFileIO::from_path(path)` / `TokioFileIO::to_path(path)`	✅ Tokio `fs`
`fromInputStream` (std)	`StreamConverters::from_reader(reader)`	✅
`fromOutputStream` (std)	`StreamConverters::to_writer(writer)`	✅
TCP	`TokioByteSink` / `TokioByteSource` / `TokioTcp`	✅ plain TCP
TLS-over-TCP	`datum_net::TokioTls`	✅ satellite crate; caller-provided rustls config
Connection lifecycle	`datum_net::ConnectionSettings` / `RetryPolicy` / `TokioTls::outgoing_connection_with_lifecycle(...)`	✅ connect timeout, TLS handshake timeout, retry, and half-close-on-upstream-finish
UDP datagrams	`datum_net::TokioUdp` / `datum_net::Udp`	✅ satellite crate; lossy bounded datagram source/sink/flow
QUIC streams	`datum_net::TokioQuic` / `datum_net::Quic`	✅ satellite crate; Quinn endpoint bind/connect plus reliable bidirectional byte streams
Framing	`Framing::delimiter(...)` / `Framing::length_field(...)`	✅
JSON framing	`Framing::json(...)`	✅
`asInputStream` / `asOutputStream` handle	`StreamConverters::as_input_stream(timeout)` / `StreamConverters::as_output_stream(timeout)`	✅ blocking `Read`/`Write` bridges; see Streaming IO guide
JVM `javaCollector*` / `fromJavaStream`	—	❌ JVM-only

Dynamic stream operators

See Dynamic Streams guide.

Akka concept	Datum operator	Notes
`KillSwitches.single`	`KillSwitches::single()`	✅ `UniqueKillSwitch`
`KillSwitches.shared`	`KillSwitches::shared(name)`	✅ `SharedKillSwitch`
`MergeHub`	`MergeHub::source(buffer_size)`	✅
`BroadcastHub`	`BroadcastHub::sink(buffer_size)`	✅
`PartitionHub`	`PartitionHub::sink(router, start_after, buffer_size)`	✅

Context propagation operators

Akka operator	Datum operator	Notes
`asSourceWithContext`	`Source::as_source_with_context(extract)`	✅
`asFlowWithContext`	`Flow::as_flow_with_context(collapse, extract)`	✅
`FlowWithContext`	`FlowWithContext<In, CtxIn, Out, CtxOut, Mat>`	✅
`SourceWithContext`	`SourceWithContext<Out, Ctx, Mat>`	✅

Testing operators

See Testing Streams guide.

Akka testkit concept	Datum equivalent	Notes
`TestSource.probe`	`datum::testkit::TestSource`	✅ manual demand + emit/complete/fail
`TestSink.probe`	`datum::testkit::TestSink`	✅ `request` / `expect_next` / `expect_complete` / `expect_error`

Graph stage API

See Working with Graphs guide and Custom Stream Processing.

Akka concept	Datum equivalent	Notes
`GraphStage`	`GraphStage<S>` trait	✅
`GraphStageLogic`	`GraphStageLogic`	✅ `pull` / `grab` / `push` / `complete` / `fail`
`InHandler` / `OutHandler`	`InHandler` / `OutHandler`	✅
Async callbacks (`getAsyncCallback`)	`GraphStageLogic::get_async_callback` / `AsyncCallback`	✅
Timers in stages	`GraphStageLogic::schedule_once` / `schedule_periodically` + `TimerHandler::on_timer`	✅
`AsyncBoundary`	`AsyncBoundary<T>::new()` graph stage; `.async_boundary()` for linear `Source` / `Flow`	✅ bounded async-boundary handoff; `.r#async()` is the raw-identifier alias
Graph buffer stage	`Buffer<T>::new(size, strategy)`	✅ graph-stage equivalent for feedback liveness breaks
Graph take-while stage	`TakeWhile<T>::new(predicate)`	✅ useful for finite feedback loops
Graph cycles	`GraphDsl` cyclic topologies	✅ fused cycles run through the queued erased interpreter; naive unbounded cycles surface the event limit

Deferred and out-of-scope summary

Category	Status
Reactive Streams interop (`asPublisher`, `asSubscriber`, `fromPublisher`, `fromSubscriber`)	❌ out of scope — Rust has no equivalent ubiquitous standard; use Tokio futures
JVM-only operators (`fromJavaStream`, `asJavaStream`, `javaCollector`, `CompletionStage`)	❌ JVM-only — capability covered by `Source::from_iterable` and `Source::future`
StreamRefs (distributed stream references)	✅ Ractor-backed `SourceRef` / `SinkRef` with `StreamRefSettings` + `StreamRefs::source_ref[_with_settings]` / `sink_ref[_with_settings]`
Graph cycles + liveness	✅ `MergePreferred` / `Broadcast` feedback loops supported; unbuffered cycles surface `EventLimitExceeded` at `FusedExecutionConfig.event_limit`
`asInputStream` / `asOutputStream` sink/source adapters	✅ `StreamConverters::as_input_stream(timeout)` / `StreamConverters::as_output_stream(timeout)`
QUIC uni-directional streams / richer connection lifecycle	⏭️ deferred

Operator Reference ​

Source operators ​

Sink operators ​

Flow operators (simple / transforming) ​

Async operators ​

Time-aware operators ​

Backpressure & rate operators ​

Substream operators ​

Fan-in operators ​

Fan-out operators ​

Actor interop operators ​

Error-handling operators ​

IO & compression operators ​

Dynamic stream operators ​

Context propagation operators ​

Testing operators ​

Graph stage API ​

Deferred and out-of-scope summary ​