Datum

Appearance

Topic — pub-sub broadcast ​

Topic<T> is Datum's stream-native, many-publisher / many-subscriber broadcast primitive. Any number of publishers push values in; any number of subscribers each receive their own independent copy of the stream. It mirrors FS2 Topic and ZIO Hub semantics, and it lives in datum-core alongside the other concurrency primitives.

The mental model ​

Think of a Topic as a live radio broadcast, not a recording.

  • When you tune in (subscribe), you hear what is broadcast from that moment on. You do not get a replay of what aired before you arrived.
  • Everyone tuned in hears the same broadcast in the same order.
  • If nobody is listening, the DJ keeps talking and the words simply vanish into the air.
  • If your receiver can't keep up with the stream, what happens next depends on the topic's overflow policy — the broadcast either waits for you, skips ahead to the latest, or drops the bits you couldn't buffer.

That "only what comes after you tune in" rule is the sharpest contrast with Signal<T>. A Signal is a dashboard gauge: the instant you subscribe, you see the current value, then updates. A Topic seeds nothing — a fresh subscriber's stream begins empty and fills only with elements published after it registered.

When to reach for Topic ​

You want…Reach forWhy
Many publishers → many independent subscriber streams, each seeing post-subscription elementsTopic<T>This page
A single upstream fanned to N consumers with Akka's slowest-consumer contractBroadcastHubAkka semantics preserved; single backpressure policy
Each subscriber to see the current value on subscribe, then updatesSignal / SubscriptionState cells seed the latest value
Many producers → one consumer (work handoff, not broadcast)ChannelMPSC, single active consumer
Coarse, possibly cross-node actor events (one message per publish per subscriber)ActorPubSubRactor pg-group pub-sub, actor-per-element

If two subscribers need the same broadcast but at different speeds and with different tolerance for falling behind, Topic is the right tool — each subscriber has its own buffer and the topic's policy decides what happens when one lags.

Why it is built this way ​

Topic follows M9's two-plane rule, and the choice is worth understanding because it explains both the performance and the exact semantics you get.

The two planes ​

  • A Ractor control-plane actor owns only the low-rate, correctness-sensitive decisions: subscribe, unsubscribe, close, terminal state, and publishing an ArcSwap snapshot of the current subscriber table.
  • The data plane is lock-free. To publish, a publisher claims one global sequence turn, loads the current subscriber-table snapshot, and enqueues an Arc<T> directly into each subscriber's lock-free ArrayQueue slot — no actor message, no mutex on the accepted path. A subscriber is woken only on an empty-to-non-empty transition or when it is parked at the head of its queue.

Why not simply send an actor message per element? ​

Because that cost is measured, and it is large. Ractor boxes every delivered message — about 848 bytes per message — so an actor-per-element broadcast would bake roughly 848 MB of allocation into every million elements, multiplied by the fan-out. The M9 write-path diagnostic confirms the round trip directly: an actor message that only enqueues and drains a no-op handler costs about 435 ns and ~999 B per transition, whereas the two-plane, caller-thread write path is about 192–270 ns and ~56 B. (Numbers from roadmap/benchmarks/concurrency-primitives.md.)

The cautionary counterexample is Akka Typed's actor Topic, which does send one message per publish per subscriber; its own documentation positions it for coarse clustered events, not stream throughput. Datum keeps the actor for what actors are good at — serializing lifecycle — and keeps it off the hot path entirely.

What the design buys you, in practice ​

The global sequence turn is what guarantees one global publish order: every subscriber that receives two elements observes them in the same order as every other subscriber, even with many concurrent publishers. The ArcSwap snapshot is what makes "subscribers see only post-subscription elements" precise: a publish acts on the subscriber table as of its linearization point, so a subscriber registered a moment later is simply not in that snapshot. And per-subscriber slot queues are what make the overflow policies per subscriber — one slow subscriber's buffer decisions never corrupt another's stream.

This also explains why Topic is a distinct type from BroadcastHub rather than an alias. BroadcastHub deliberately preserves Akka's semantics (single slowest-consumer backpressure contract, fed by one materialized upstream sink). Topic is the FS2/ZIO-modeled primitive: arbitrary publishers, explicit per-subscriber overflow policy, FS2 zero-subscriber drop. The two coexist; neither changed the other.

Examples ​

All snippets below are imported from a cargo test-verified integration test, so they compile and their assertions hold.

Fan-out and global order ​

Create a topic, subscribe two consumers, publish. Each subscriber receives every element published after it registered, in the same order.

rust
use datum::{Topic, TopicOverflow, testkit::TestSink};

// Per-subscriber capacity 8, correctness-first Backpressure policy.
let topic = Topic::new(8, TopicOverflow::Backpressure).unwrap();

// Each `subscribe()` is a Source blueprint; materializing it registers a
// fresh subscriber. Registration is synchronous, so both slots are live
// before we publish.
let left = topic.subscribe().run_with(TestSink::probe()).unwrap();
let right = topic.subscribe().run_with(TestSink::probe()).unwrap();
assert_eq!(topic.subscriber_count(), 2);

// A single publisher thread claims sequence turns 1, 2, 3 in order.
topic.try_publish("alpha").unwrap();
topic.try_publish("beta").unwrap();
topic.try_publish("gamma").unwrap();

// Both subscribers observe the *same* global order.
left.request(3);
right.request(3);
left.assert_next_n(["alpha", "beta", "gamma"]);
right.assert_next_n(["alpha", "beta", "gamma"]);

// Closing lets current subscribers drain, then completes them.
topic.close().unwrap();
left.request(1);
right.request(1);
left.expect_complete();
right.expect_complete();

Topic::new(capacity, overflow) sets the per-subscriber buffer capacity (which must be greater than zero) and the overflow policy. The handle is Clone — clone it to hand out to publishers. subscribe() returns an ordinary Source<T> blueprint (T: Clone); registration happens synchronously at materialization, which is why subscriber_count() already reads 2 before the first publish.

Only post-subscription elements — and the zero-subscriber drop ​

A Topic seeds nothing. A publish that happens while there are no subscribers is accepted and dropped; there is no replay buffer for future subscribers. This is FS2 Topic behavior, not ZIO Hub's buffering-for-future-subscribers.

rust
use datum::{Topic, TopicOverflow, testkit::TestSink};

let topic = Topic::new(8, TopicOverflow::Backpressure).unwrap();

// No subscribers yet: this publish is accepted and dropped. There is no
// replay buffer for future subscribers (FS2 `Topic`, not ZIO `Hub`).
topic.try_publish(1_u64).unwrap();

let sub = topic.subscribe().run_with(TestSink::probe()).unwrap();

// Only elements published *after* this subscription are delivered.
topic.try_publish(2).unwrap();
topic.try_publish(3).unwrap();
topic.close().unwrap();

sub.request(3);
sub.assert_next_n([2, 3]); // `1` was published to zero subscribers and dropped
sub.expect_complete();

Overflow policies ​

When a subscriber's buffer is full, the policy decides what gives. Sliding and Dropping never make a publisher wait; they trade completeness for liveness in opposite directions.

rust
use datum::{Topic, TopicOverflow, testkit::TestSink};

// Sliding: a full subscriber drops its OLDEST queued element to make room.
let sliding = Topic::new(2, TopicOverflow::Sliding).unwrap();
let sub = sliding.subscribe().run_with(TestSink::probe()).unwrap();
// The subscriber has not requested yet, so its capacity-2 buffer fills up.
sliding.try_publish(1_u64).unwrap();
sliding.try_publish(2).unwrap();
sliding.try_publish(3).unwrap(); // buffer full: 1 slides out
sliding.try_publish(4).unwrap(); // buffer full: 2 slides out
sliding.close().unwrap();
sub.request(2);
sub.assert_next_n([3, 4]); // only the newest window survived
sub.request(1);
sub.expect_complete();

// Dropping: a full subscriber skips the NEW element instead.
let dropping = Topic::new(2, TopicOverflow::Dropping).unwrap();
let sub = dropping.subscribe().run_with(TestSink::probe()).unwrap();
dropping.try_publish(1_u64).unwrap();
dropping.try_publish(2).unwrap();
dropping.try_publish(3).unwrap(); // buffer full: 3 is skipped
dropping.try_publish(4).unwrap(); // buffer full: 4 is skipped
dropping.close().unwrap();
sub.request(2);
sub.assert_next_n([1, 2]); // only the oldest window survived
sub.request(1);
sub.expect_complete();

Fanning one ingest into independent pipelines ​

Because subscribe() is a normal Source, each subscriber can carry its own operators and terminate in its own Sink. The two pipelines below run and backpressure independently. This snippet uses the prelude glob instead of explicit imports:

rust
use datum::prelude::*;

let topic = Topic::new(16, TopicOverflow::Backpressure).unwrap();

// Each subscriber is an ordinary Source: attach any operators and any Sink.
// These two pipelines run and backpressure independently of each other.
let doubled = topic
    .subscribe()
    .map(|n| n * 2)
    .run_with(Sink::collect())
    .unwrap();
let evens = topic
    .subscribe()
    .filter(|n| n % 2 == 0)
    .run_with(Sink::collect())
    .unwrap();
assert_eq!(topic.subscriber_count(), 2);

// Drive the ingest. A single publisher fixes the global order 1, 2, 3, 4.
for value in 1_u64..=4 {
    topic.try_publish(value).unwrap();
}
topic.close().unwrap();

assert_eq!(doubled.wait().unwrap(), vec![2, 4, 6, 8]);
assert_eq!(evens.wait().unwrap(), vec![2, 4]);

Using it properly ​

Choosing an overflow policy ​

The policy is fixed at construction and applies to every subscriber (each with its own buffer). Choose by what a lagging subscriber should cost you:

PolicyFull-buffer behavior (per subscriber)Publisher effectReach for it when
BackpressureNothing is lost; the publish waits for roomSlowest subscriber gates all publishersCorrectness first — every subscriber must see every element
SlidingDrops that subscriber's oldest queued element, enqueues the new onePublishers never waitLatest-window telemetry — a lagging dashboard should jump to now
DroppingSkips the new element for that subscriber onlyPublishers never waitAdmission control — shed load, keep the earliest accepted work

The load-bearing detail for Backpressure: because publishers claim a strict global sequence turn, a publisher stalled on one full, slow subscriber holds its turn and therefore backpressures every other publisher too. That is the intended correctness-first contract, but it means a single slow subscriber can throttle the whole topic. Sliding and Dropping never wait for capacity, so a slow subscriber under those policies degrades only its own stream.

publish vs try_publish ​

  • publish(value).await waits only under Backpressure (for its sequence turn and for capacity). Under Sliding/Dropping it enqueues and returns without ever awaiting a subscriber. It returns Err(TopicPublishError::Closed(value)) if the topic is closed, handing the value back.
  • try_publish(value) never blocks. It can return TopicTryPublishError::Full(value) (a Backpressure subscriber has no room), ::Busy(value) (another publisher currently owns an earlier global turn), or ::Closed(value). In each case your value is returned to you.

A single publisher thread issuing sequential try_publish calls never sees Busy — each call finishes its turn before the next begins. Busy is a concurrency signal, not an error to fear.

Close semantics ​

close() is graceful: current subscribers drain their queued elements first, then complete. Publishers that begin after close fail with Closed. Subscribing after close yields a source that completes immediately (empty). is_closed() polls the state; closed().await waits for it.

Pitfalls ​

  • A subscriber that never pulls fills its buffer. The buffer sits between the lock-free slot and your downstream. Under Backpressure a non-consuming subscriber will stall publishers; under Sliding/Dropping it will silently lose elements. This is exactly what the overflow snippet relies on to be deterministic — make sure it's what you want in production.
  • Dropping every topic handle without close() is not graceful. The control actor stops when the last Topic clone is dropped, and its subscribers then fail with StreamError::ActorTerminated rather than completing. Call close() (or keep a handle alive) if you want clean completion.
  • T is cloned per subscriber, per element. Internally each value is an Arc<T>, but the subscriber source hands you an owned T via Clone. For large payloads, prefer Topic<Arc<Big>> so the per-subscriber clone is just a refcount bump.
  • Nothing is replayed. If a subscriber must not miss the first elements, subscribe (and confirm subscriber_count()) before the first publish — as every example here does.
  • capacity == 0 panics. Capacity is per subscriber and must be positive.

Integration tips ​

Topic composes with the rest of Datum because a subscription is just a Source, and publishing is just a method call you can make from anywhere.

  • Streams and graphs. topic.subscribe() is a Source<T> — attach any Flow, feed it into a GraphDsl junction, or terminate it in any Sink. The integration snippet above is the pattern: one ingest fanned to several independently-shaped, independently-backpressured consumer pipelines.
  • Different policies for different consumers. A single topic has one policy. When one consumer must be lossless while another only wants the latest window, feed two topics (e.g. one Backpressure, one Sliding) from the same ingest source — publish each element to both. Each subscriber set then gets the contract it needs.
  • An event bus between actors and streams. Clone the Topic handle into a Ractor actor and call try_publish(event) from its message handler; stream pipelines elsewhere subscribe and process at their own pace. Because publishing is lock-free and off the actor's own hot path, the bus does not become an actor-mailbox bottleneck. See Actor interop for the actor side.
  • Metrics / telemetry fan-out. A Sliding topic is a natural telemetry bus: fast producers publish continuously, and each dashboard/exporter subscriber keeps only the newest window if it falls behind, never stalling the producers.
  • KillSwitches. Each subscriber pipeline is an ordinary stream, so you can splice a KillSwitch into one consumer to tear it down independently of the others; dropping a subscriber's stream also unsubscribes it and frees its slot.
  • datum-net. A subscriber pipeline can terminate in a TCP/TLS sink to broadcast a topic across a process boundary — see datum-net.

Performance ​

On the same-host M9 comparison, Topic beats the best JVM competitor per row on both wall-clock and CPU. The fan-out target is ZIO Hub; the overflow target is ZIO Hub's dropping policy (FS2 Topic and Akka BroadcastHub have no direct sliding/dropping topic policy, so no rows are fabricated for them).

ScenarioSweepBest JVM (wall)Datum wallDatum CPU vs best JVM
topic_fanout_1024xN1 subZIO Hub 430 µs/op108 µs/op (3.98× faster)1.44× lower
topic_fanout_1024xN64 subsZIO Hub 89,765 µs/op2,608 µs/op (34.42× faster)237.77× lower
topic_fanout_1024xN1,024 subsZIO Hub 1,430,782 µs/op61,891 µs/op (23.12× faster)157.64× lower
topic_overflow (sliding/dropping)1–64 subsZIO Hub.dropping1.23–21.33× faster wall3.51–7.02× lower

Across the full fan-out sweep Datum is 3.98–34.42× faster wall and 1.44–237.77× lower CPU than ZIO Hub. These are honest whole-process numbers, including CPU — a path is not allowed to win wall-clock by busy-spinning. The full record, with p50/p99, allocation, peak RSS, correctness counters, and the per-round implementation notes, is in roadmap/benchmarks/concurrency-primitives.md.

See also ​