Appearance
Migrating from Akka Streams
This guide is for Scala/Akka Streams users coming to Datum. It covers the name map, the behavioral differences that matter most, and what is intentionally absent.
Name map
Imports.
use datum::prelude::*;brings the common linear and graph types (Source,Flow,Sink,GraphDsl, junctions, shapes, …) into scope in one line. Importing individual types explicitly still works exactly as before — the prelude is a convenience, not a requirement.
Source constructors
| Akka (Scala) | Datum (Rust) |
|---|---|
Source.empty | Source::empty() |
Source.single(x) | Source::single(x) |
Source.repeat(x) | Source::repeat(x) |
Source.failed(ex) | Source::failed(error) |
Source.fromIterator(() => iter) | Source::from_fn_iter(|| iter) |
Source(iterable) / Source.from(iterable) | Source::from_iterable(items); for Vec/arrays also Source::from(vec) / vec.into() |
Source(1 to 100) / Source.range(1, 100) | Source::from_iterable(1..=100) or (1..=100).into_source() |
Source.future(f) | Source::future(factory) |
Source.futureSource(f) | Source::future_source(factory) |
Source.maybe | Source::maybe() — returns (MaybeHandle<T>, Source<T>) |
Source.never | Source::never() |
Source.tick(init, interval, x) | Source::tick(init, interval, x) |
Source.cycle(() => iter) | Source::cycle(factory) |
Source.unfold(s)(f) | Source::unfold(s, f) |
Source.unfoldAsync(s)(f) | Source::unfold_async(s, f) |
Source.unfoldResource(create, read, close) | Source::unfold_resource(create, read, close) |
Source.unfoldResourceAsync(...) | Source::unfold_resource_async(...) |
Source.lazily(() => src) | Source::lazy_source(factory) |
Source.lazyFuture(() => fut) | Source::lazy_future(factory) |
Source.queue() | Source::queue(capacity, strategy) — SourceQueue<T> is the materialized value; Source::queue_bounded(capacity) for BoundedSourceQueue<T> |
Source.asSourceWithContext[Ctx](extract) | source.as_source_with_context(extract) |
Source.combine(s1, s2, rest: _*)(merge) | Source::combine(sources, strategy) |
Sink constructors
| Akka (Scala) | Datum (Rust) |
|---|---|
Sink.ignore | Sink::ignore() |
Sink.head | Sink::head() |
Sink.headOption | Sink::head_option() |
Sink.last | Sink::last() |
Sink.lastOption | Sink::last_option() |
Sink.seq | Sink::collect() — returns Vec<T> |
Sink.takeLast(n) | Sink::take_last(n) |
Sink.fold(z)(f) | Sink::fold(zero, f) |
Sink.reduce(f) | Sink::reduce(f) |
Sink.foreach(f) | Sink::foreach(f) |
Sink.cancelled | Sink::cancelled() |
Sink.onComplete(f) | Sink::on_complete(f) |
Sink.fromMaterializer | Sink::from_materializer(factory) |
Sink.queue() | Sink::queue() — SinkQueue<T> is the materialized value |
Sink.combine(s1, s2)(strategy) | Sink::combine(sinks, strategy) |
StreamConverters.asInputStream(timeout) | StreamConverters::as_input_stream(read_timeout) — materializes InputStreamHandle (impl Read) |
StreamConverters.asOutputStream(timeout) | StreamConverters::as_output_stream(write_timeout) — materializes OutputStreamHandle (impl Write) |
Running a stream
Akka's runWith / runFold / runForeach / runReduce map to terminals on Source. The run_* shortcuts are thin wrappers over run_with(Sink::…) and keep the sink's materialized value — both styles are available; use whichever reads best.
| Akka (Scala) | Datum (Rust) |
|---|---|
src.runWith(sink) | source.run_with(sink) |
src.runFold(z)(f) | source.run_fold(z, f) — or source.run_with(Sink::fold(z, f)) |
src.runForeach(f) | source.run_foreach(f) / source.run_for_each(f) |
src.runReduce(f) | source.run_reduce(f) |
src.runWith(Sink.seq) | source.run_collect() |
Flow operators
| Akka (Scala) | Datum (Rust) | Notes |
|---|---|---|
.map(f) | .map(f) | |
.filter(p) | .filter(p) | |
.filterNot(p) | .filter_not(p) | |
.collect { case ... } | .filter_map(f) | Rust rename — collect conflicts with Iterator::collect |
.mapConcat(f) | .map_concat(f) | |
.grouped(n) | .grouped(n) | |
.sliding(n, step) | .sliding(n, step) | |
.scan(z)(f) | .scan(z, f) | |
.fold(z)(f) | .fold(z, f) | |
.reduce(f) | .reduce(f) | |
.take(n) | .take(n) | |
.takeWhile(p) | .take_while(p) | |
.drop(n) | .drop(n) | |
.dropWhile(p) | .drop_while(p) | |
.limit(n) | .limit(n) | |
.statefulMap(factory)(f) | .stateful_map(seed, f) | |
.statefulMapConcat(factory)(f) | .stateful_map_concat(seed, f) | |
.mapAsync(p)(f) | .map_async(p, f) | |
.mapAsyncUnordered(p)(f) | .map_async_unordered(p, f) | |
.mapAsyncPartitioned(p)(part)(f) | .map_async_partitioned(parallelism, per_partition, partition_fn, f) | |
.intersperse(inject) | .intersperse(inject) | |
.groupedWeighted(max)(cost) | .grouped_weighted(max_weight, cost_fn) | |
.limitWeighted(max)(cost) | .limit_weighted(max_weight, cost_fn) | |
.contramap(f) | .contramap(f) | |
.log(name) | .monitor(callback) | Datum uses a closure, not a marker/logging framework |
.monitor | .monitor(callback) | |
.watchTermination()(f) | .watch_termination(materialize_callback) | |
.foldAsync(z)(f) | .fold_async(z, f) | |
.scanAsync(z)(f) | .scan_async(z, f) | |
.mapWithResource(create)(f, close) | .map_with_resource(create, f, close) | |
.throttle(elements, per, burst, mode) | .throttle(elements, per, burst, mode) | |
.delay(d) | .delay(d, strategy) | DelayOverflowStrategy required |
.initialDelay(d) | .initial_delay(d) | |
.groupedWithin(n, d) | .grouped_within(n, d) | |
.takeWithin(d) | .take_within(d) | |
.dropWithin(d) | .drop_within(d) | |
.idleTimeout(d) | .idle_timeout(d) | |
.backpressureTimeout(d) | .backpressure_timeout(d) | |
.completionTimeout(d) | .completion_timeout(d) | |
.initialTimeout(d) | .initial_timeout(d) | |
.keepAlive(d, inject) | .keep_alive(d, inject) | |
.buffer(n, strategy) | .buffer(n, strategy) | OverflowStrategy |
.conflate(f) | .conflate(f) | |
.conflateWithSeed(seed)(f) | .conflate_with_seed(seed, f) | |
.expand(f) | .expand(f) | |
.extrapolate(f) | .extrapolate(f, initial) | |
.batch(max, seed)(agg) | .batch(max, seed, aggregate) | |
.batchWeighted(max, cost)(seed)(agg) | .batch_weighted(max, cost_fn, seed, aggregate) | |
.recover { case ... } | .recover(f) | |
.recoverWith { case ... } | .recover_with(f) | |
.recoverWithRetries(n) { case ... } | .recover_with_retries(n, f) | |
.mapError(f) | .map_error(f) | |
RestartSource.withBackoff(settings)(() => src) | RestartSource::with_backoff(settings, factory) | |
RestartFlow.withBackoff(settings)(() => flow) | RestartFlow::with_backoff(settings, factory) | |
RetryFlow.withBackoff(...) | RetryFlow::with_backoff(min, max, factor, f) | |
.flatMapConcat(f) | .flat_map_concat(f) | |
.flatMapMerge(breadth, f) | .flat_map_merge(breadth, f) | |
.prefixAndTail(n) | .prefix_and_tail(n) | emits (Vec<T>, Source<T>) |
.flatMapPrefix(n)(f) | .flat_map_prefix(n, f) | |
.groupBy(max, key) | .group_by(max_substreams, key_fn, allow_recreation) | last arg toggles whether a closed key re-opens a fresh substream |
.splitWhen(p) | .split_when(p) | emits Source<T> per segment |
.splitAfter(p) | .split_after(p) | emits Source<T> per segment |
.concat(src) | .concat(src) | |
.prepend(src) | .prepend(src) | |
.orElse(src) | .or_else(src) | |
.interleave(src, n) | .interleave(src, n) | |
.mergeSorted(src) | .merge_sorted(src) | |
.mergeLatest(src) | .merge_latest(src, eager_complete) | |
.mergeAll(srcs, eager) | .merge_all(srcs, eager_complete) | |
.zipWith(src)(f) | .zip_with(src, f) | |
.zipLatest(src) | .zip_latest(src) | |
.zipLatestWith(src)(f) | .zip_latest_with(src, f) | |
.zipWithIndex | .zip_with_index() | emits (T, u64) |
.zipAll(src, thisZ, thatZ) | .zip_all(src, this_default, that_default) | |
.alsoTo(sink) | .also_to(sink) | |
.alsoToAll(sinks) | .also_to_all(sinks) | |
.divertTo(sink, p) | .divert_to(sink, predicate) | |
.wireTap(sink) | .wire_tap(sink) |
Actor interop
| Akka (Scala) | Datum (Rust) |
|---|---|
ActorFlow.ask(ref, timeout)(f) | ActorFlow::ask(actor_ref, parallelism, timeout, make_msg) |
ActorFlow.askWithStatus(ref, timeout)(f) | ActorFlow::ask_with_status(actor_ref, parallelism, timeout, make_msg) |
ActorFlow.askWithContext(ref, timeout)(f) | ActorFlow::ask_with_context(actor_ref, parallelism, timeout, make_msg) |
Source.actorRef(bufferSize, overflowStrategy) | ActorSource::actor_ref() |
Sink.actorRef(ref, onCompleteMsg) | ActorSink::actor_ref(...) |
StreamRefs.sourceRef() | StreamRefs::source_ref() |
StreamRefs.sinkRef() | StreamRefs::sink_ref() |
Concurrency primitives
Datum also ships FS2/ZIO-class stream-native concurrency primitives. They are not Akka Streams names, but they cover common migration surfaces when Akka applications also use FS2 or ZIO around stream boundaries.
| FS2 / ZIO | Datum | Notes |
|---|---|---|
FS2 Channel | Channel<T> / Source::channel(capacity) | Closeable bounded MPSC handoff; many producers, one consumer stream |
FS2 Topic | Topic<T> | Pub-sub broadcast; per-subscriber Backpressure, Sliding, or Dropping overflow |
FS2 SignallingRef | Signal<T> | Latest-value state with synchronous get() / get_cloned() and a coalesced changes() feed |
ZIO Hub | Topic<T> | Many publishers and many subscribers; slow-subscriber behavior is selected with TopicOverflow |
ZIO SubscriptionRef | Subscription<T> | Latest-value state plus a bounded every-change feed; use SubscriptionOverflow::Backpressure for lossless delivery |
ZIO Queue | Channel<T> | Bounded MPSC queue feeding a Datum Source<T> |
See Concurrency Primitives for examples and performance notes.
Behavioral differences
Failures flow through Result, not exceptions
In Akka, user callbacks throw exceptions to signal failure, and supervision strategies intercept Throwable. In Datum, failures are explicit: a failed stream emits a StreamError and terminates. Operators that apply user closures have explicit fallible variants (e.g. try_map, try_filter_map, try_fold) that return Result<T, StreamError>. Non-fallible variants (map, filter, etc.) stay branch-free and are not retroactively supervised. The older unsupervised _result names remain deprecated aliases for source compatibility.
When a callback can fail and should be supervised, use the explicit *_result_with_supervision variant and pass a SupervisionDecider:
rust
use datum::{Supervision, StreamError};
source.map_result_with_supervision(
|x| if x == 0 { Err(StreamError::Failed("zero".into())) } else { Ok(x) },
Supervision::resuming_decider(),
)SupervisionDirective::Stop (fail the stream), Resume (drop element), and Restart (drop and reset state) work the same as Akka's supervision strategies.
Graph wiring strictness is split by construction mode
Akka's GraphDSL encodes legal port connections in the type system, so many illegal wiring mistakes are compile errors. Datum now has two Rust graph-construction modes:
GraphDsl::typed_try_createusesTypedGraphBuilder, whose move-only open-port tokens make double-wiring a compile error for the static subset. Typedconnectalso rejectsOutlet<T> -> Inlet<U>mismatches at compile time.GraphDsl::try_createkeeps the mutableGraphBuilderfor dynamic topologies, cycles, partial graph imports, method-basedwire, and explicit erased interop throughconnect_any. Shape completeness and erased wiring are still runtime-validated and returnStreamResult.
For mutable-builder wiring syntax, Datum offers two equivalent, fully-supported styles. Akka's bcast ~> merge maps to builder.wire(bcast.to(&merge)) (auto port selection, chainable, errors deferred to graph creation), and the explicit builder.connect(bcast.outlet(0)?, merge.inlet(0)?)? remains available for exact ports and per-edge error handling. connect_any is the visible runtime-typed interop path. MergePreferred and Bidi shapes are explicit-only in the wire DSL. See Working with Graphs.
Blueprint vs materialization (same contract, explicit lifetime)
Both Akka and Datum distinguish between building a blueprint and running it. In Datum the distinction is more explicit: construction methods (Source::from_iterable, .map(...), etc.) have no side effects and start nothing. Execution begins only at .run() / .run_with() / materializer.materialize(graph), which return materialized handles immediately. See the Blueprint vs Run concept page for details.
Tokio-first async; no Reactive Streams interop
Datum is built on Tokio. Ractor (the actor runtime) runs on Tokio, and async operators like map_async dispatch futures onto the Tokio runtime. std::future::Future is accepted as an optional portability surface, but Tokio is the default and the primary tested path.
Reactive Streams publisher/subscriber interop (asPublisher, asSubscriber, fromPublisher, fromSubscriber) is intentionally absent. Rust has no equivalent ubiquitous standard; the idiomatic bridge is Tokio's Stream/Sink traits. See Reactive Streams interop below.
SubFlow vs nested Source
Akka's groupBy / splitWhen / splitAfter return a SubFlow — a specialized type with operators that apply to each substream. Datum returns a Source<Source<T>>: each substream is a plain Source<T> you can operate on directly. This means SubFlow-specific methods like .mergeSubstreams() and .concatSubstreams() do not exist as named methods; instead, use flat_map_merge(breadth, |sub| sub) or flat_map_concat(|sub| sub).
group_by key retention
Akka's groupBy tracks cancelled/closed keys until the substream completes (i.e. a re-opened key gets its own fresh substream). Datum mirrors this: group_by(max, key_fn, false) retains closed keys until the parent stream completes. High-cardinality churn (many distinct keys that each emit one element) can therefore grow memory linearly in the number of distinct closed keys seen.
StreamRefs
StreamRefs are available as Ractor-backed one-shot handles: StreamRefs::source_ref() materializes SourceRef<T>, and StreamRefs::sink_ref() materializes SinkRef<T>. Endpoints can be configured via StreamRefSettings (buffer capacity, subscription timeout, demand redelivery interval). Same-process refs splice directly for the fast local path. Cross-process transport is shipped: datum-core owns a transport-agnostic StreamRefs protobuf protocol, and the datum-net satellite carries the frames over plaintext TCP or TLS-encrypted QUIC. See the StreamRefs guide for the remote API and the honest forced-remote benchmark.
Graph cycles
GraphBuilder accepts cyclic fused graphs. MergePreferred / Broadcast feedback loops and bounded feedback paths run through Datum's queued erased interpreter. As in Akka, MergePreferred can keep feedback elements circulating while starving secondary inputs if nothing exits the loop; use Buffer<T>::new(...), dropping/backpressure strategy choices, or a closing stage such as TakeWhile<T>::new(...) when the loop must be productive and finite. A naive unbounded Merge + Broadcast loop still has Akka's liveness problem: it cannot recover buffer space. Datum surfaces that as StreamError::EventLimitExceeded when the configured fused event budget (FusedExecutionConfig::event_limit, default 100M) is exhausted, instead of silently hanging.
IO adapters (asInputStream / asOutputStream)
StreamConverters::as_input_stream(read_timeout) is a Sink<Vec<u8>> that materializes an InputStreamHandle (implements std::io::Read). StreamConverters::as_output_stream(write_timeout) is a Source<Vec<u8>> that materializes an OutputStreamHandle (implements std::io::Write). These mirror Akka's StreamConverters.asInputStream / asOutputStream semantics: the sink feeds the read handle, the write handle feeds the source, and cancellation/close is explicit.
Networking
Plain TCP lives in datum-core as TokioTcp. The datum-net satellite covers the richer Akka IO networking cases: Akka classic IO Tcp/TLS maps to datum_net::Tls and datum_net::TokioTls, Akka Udp maps to datum_net::Udp and datum_net::TokioUdp, and QUIC streams use datum_net::Quic and datum_net::TokioQuic. TLS client connection lifecycle controls such as connect timeout, handshake timeout, retry, and half-close-on-upstream-finish are configured with datum_net::ConnectionSettings and datum_net::RetryPolicy.
Cluster and sharding
Akka Cluster and Cluster Sharding map onto the v0.10 satellite crates datum-cluster (membership, downing, placement) and datum-cluster-sharding (entities, rebalance, passivation). The membership state names are deliberately identical; the substrate underneath is SWIM gossip (foca), not Akka's Artery/gossip stack, and stays entirely private.
Akka Cluster → datum-cluster
| Akka (Scala) | Datum (Rust) | Notes |
|---|---|---|
Cluster(system) | ClusterNode::start(ClusterConfig) | or ClusterAgent::start for the cluster-aware agent (membership + DCP + registry) |
cluster.subscribe(…, classOf[MemberEvent]) | node.events() → Subscription<MemberEvent> | lossless, ordered, backpressuring |
cluster.state / registerOnMemberUp | node.state() → Signal<ClusterState>; node.current_state() | latest snapshot, no replay |
MemberStatus.{Joining,Up,Leaving,Exiting,Down,Removed} | MemberState::{Joining,Up,Leaving,Exiting,Down,Removed} | same names; reachability is a separate member.unreachable flag |
seed nodes (akka.cluster.seed-nodes) | ClusterConfig::with_seed_nodes([addr, …]) | addresses only; identities are discovered via gossip |
akka.cluster.roles / cluster.selfRoles | ClusterConfig::with_roles([…]), Member::has_role | |
cluster.leave(address) | node.leave().await | walks Leaving → Exiting → Removed, gossiped |
Downing provider / SplitBrainResolver | DowningProvider trait; TimeoutDowning | no split-brain resolver yet — a quorum/lease provider is a named v0.11+ follow-up |
| Cluster Singleton | ClusterState::placement_coordinator() (oldest Up member) | a deterministic convention, not fenced during a partition |
| Distributed Data / CRDTs | (absent) | out of scope |
Akka Cluster Sharding → datum-cluster-sharding
| Akka (Scala) | Datum (Rust) | Notes |
|---|---|---|
ClusterSharding(system).init(Entity(TypeKey)(create)) | Sharding::init(&agent, ShardingConfig) + handle.register_entity_type(name, factory) | one region per node |
EntityTypeKey[M] | the string type_name + EntityRef<M> | |
ShardingEnvelope(entityId, msg) | ShardEnvelope { entity_id, message } | |
HashCodeMessageExtractor / message & shard extractors | ShardExtractor<M>; default DefaultShardExtractor | stable FNV-1a hash of entity_id modulo num_shards |
entityRefFor(TypeKey, id) | handle.entity_ref(type_name, id) | entity_ref_with_extractor for a custom extractor |
entityRef ! msg | entity_ref.tell(msg).await | at-most-once per attempt |
entityRef.ask(replyTo => msg) | entity_ref.ask(timeout, make_msg).await | make_msg: FnOnce(ReplyPort<R>) -> M |
context.self ! Passivate / passivation | entity_ref.passivate().await; ShardingConfig::passivation_idle_timeout | next message respawns |
rememberEntities = on + state store | RememberEntitiesConfig::with_store + register_remembered_entity_type; SPI RememberEntitiesStore (InMemoryStore/FileStore) | re-spawns ids, not state; writes are write-behind |
ShardAllocationStrategy / rebalance | coordinator least-shards + RebalanceReason::{DeadOwner, GracefulSpread} | GracefulSpread capped by rebalance_per_round |
| ShardCoordinator (ddata/persistence backed) | oldest-member coordinator, in-memory allocation table | no persistent allocation store in v0.10 |
EventSourcedBehavior durable entity state | (absent) | entity state is in-memory and not migrated on rebalance |
Two differences dominate a migration. First, delivery is at-most-once per attempt with caller-driven retries — design entity handlers to be idempotent or duplicate-tolerant. Second, there is no persistence in v0.10: the allocation table and entity state are in-memory, so plan around failover rebuilding state rather than restoring it. See the sharding delivery semantics for the full contract.
What is intentionally absent
Reactive Streams interop is out of scope
asPublisher, asSubscriber, fromPublisher, fromSubscriber, and the JDK Flow.Publisher / Flow.Subscriber adapters exist in Akka to bridge the JVM Reactive Streams ecosystem and its TCK. Rust has no equivalent ubiquitous standard. The idiomatic integration surface is Tokio's async Stream and Sink traits, which Datum already uses internally.
Mirroring the RS publisher/subscriber protocol would add a foreign abstraction and TCK-conformance burden without serving a Rust audience. This is a firm design decision (out of scope, not deferred).
JVM-specific operators
fromJavaStream, asJavaStream, javaCollector*, and the CompletionStage family are JVM API adapters with no meaningful Rust equivalent. The underlying capability is covered:
Source.fromJavaStream(() => javaStream)→Source::from_iterable(iter)Source.fromCompletionStage(cs)→Source::future(|| async { ... })
SubFlow surface
Datum does not expose a dedicated SubFlow type. Use nested Source<Source<T>> and the flat_map_* family to process substream results.
Quick-start conversion example
Akka (Scala):
scala
Source(1 to 10)
.map(_ * 2)
.filter(_ % 3 != 0)
.runWith(Sink.seq)Datum (Rust):
rust
use datum::Source;
let result: Vec<u64> = Source::from_iterable(1_u64..=10)
.map(|x| x * 2)
.filter(|x| x % 3 != 0)
.run_collect()?;Both return a Vec<u64>. run_collect() is the runWith(Sink.seq) shorthand — it blocks until the stream completes and returns the collected vector (equivalent to run_with(Sink::collect())?.wait()?). Note that run_with itself returns a StreamCompletion<T> handle immediately, not the value; call .wait() to block for the result.
For a running graph with a materializer:
Akka (Scala):
scala
val mat = ActorMaterializer()
val (killSwitch, done) = source
.viaMat(KillSwitches.single)(Keep.right)
.toMat(sink)(Keep.both)
.run()(mat)Datum (Rust):
rust
use datum::{KillSwitches, Runtime, Source, Sink, Keep};
let materializer = Runtime::new();
let (kill_switch, completion) = Source::repeat(1_u64)
.via_mat(KillSwitches::single(), Keep::right)
.to_mat(Sink::ignore(), Keep::both)
.run_with_materializer(&materializer)?;Keep::left, Keep::right, Keep::both, and Keep::none work identically to Akka's Keep.