Channel

Channel<T> is a closeable, bounded, many-producer / single-consumer (MPSC) handoff. Many producers send values into a shared buffer; exactly one materialized consumer stream drains them.

Concept & when to use

Picture a kitchen order rail. Several waiters (producers) clip tickets onto the rail; one cook (the consumer) works them off in order. The rail only holds so many tickets — when it is full a waiter waits for the cook to clear one before clipping the next (backpressure). At the end of service someone flips the "kitchen closed" sign: no new tickets are accepted, but the cook still finishes every ticket already on the rail before going home (graceful close).

That is exactly Channel<T>:

• Many producers → one consumer. Clone the handle and hand a clone to each producer. The consumer side is an ordinary Datum Source, so it composes with every operator, sink, and graph you already know.
• Bounded buffer. Capacity is fixed at construction and must be greater than zero.
• Backpressure or immediate-failure, your choice. send().await waits for room; try_send() never waits and tells you if the buffer is full.
• Graceful close. close() stops new sends but lets buffered elements drain, then the source completes cleanly.

When to reach for it

Reach for Channel<T> when the producer side is not naturally a Datum stream — a callback, an event listener, a thread doing blocking I/O, an FFI boundary — and you want to funnel those values into one stream pipeline with real backpressure and a clean shutdown.

It sits alongside two primitives you may already know. The distinctions are worth internalizing:

Primitive	Producer shape	Backpressure	On overflow	Close semantics
Channel<T>	Direct send/try_send calls from any code	send().await parks for room; try_send returns Full	Never drops silently — you wait or you get the value back	close() drains, then completes
SourceQueue<T> (see operators reference)	A single materialized offer handle	A blocking offer with at most one pending offer	Akka OverflowStrategy (may drop, backpressure, or fail)	Complete the queue handle
MergeHub (see Dynamic Streams)	Producers must be Datum streams attached to a materialized Sink	Per-producer stream backpressure	Backpressures upstream	Draining control over attached producers

Rules of thumb:

• Producers are already Sources? Use MergeHub — it is built for fanning streams into one.
• Producer is external code calling in, and you can tolerate OverflowStrategy drops or need Akka parity? SourceQueue may fit.
• Producer is external code calling in, you want a genuine multi-producer send with real backpressure and a graceful drain-then-complete close? Channel<T> is the one built for that gap. SourceQueue's offer is single-pending, and BoundedSourceQueue drops on full rather than backpressuring — neither is a general multi-producer send.

For a side-by-side of the whole M9 family (Signal, Subscription, Channel, Topic), see the Concurrency Primitives overview.

Design thinking

Channel<T> follows the M9 two-plane rule, and the shape of its public API falls directly out of a few deliberate maintainer decisions. Understanding them explains what the type does — and does not — promise.

Two planes: lock-free data path, control-plane close

Datum's concurrency primitives split into an actor control plane (low-rate lifecycle decisions that need serialization) and a lock-free data plane (the per-element hot path). For Channel, the data path is a crossbeam ArrayQueue ring with an atomic permit count and a waker protocol — a successful send / try_send takes no mutex and sends no actor message. Only the close/terminal state lives in the control plane.

Why not just make the channel an actor and message each element to it? Because Ractor boxes every delivered message — ~848 B per element, measured, and upstream-bound (M9 record). Baking an actor hop into the hottest path would cost ~848 MB of allocation per million elements before you moved any real work. So the element path stays lock-free; the actor is reserved for the things that are genuinely rare — and correctness-critical to serialize. This is the same design every high-throughput reference implementation converges on (ZIO Queue, Akka's hub GraphStages, FS2's Ref+Deferred machines).

send (backpressured) vs try_send (immediate) — and choosing

The two entry points exist because the two callers are genuinely different:

• send(value).await is for producers that can wait. When the ring is full the future parks through a waker protocol — it does not busy-spin — and resumes when the consumer frees a slot. Returns Err(ChannelSendError::Closed(value)) only if the channel closes before the value is accepted, handing your value back so nothing is silently lost.
• try_send(value) is for producers that must not block — a synchronous callback, a UI thread, a real-time path. It returns immediately: Ok(()), Err(TrySendError::Full(value)) if the ring is full right now, or Err(TrySendError::Closed(value)) after close. Both error variants return the rejected value so you decide the policy (retry, drop, log, spill).

Choose send when the producer's job is to feed the pipeline and it is fine for it to slow down under load — that is backpressure doing its job. Choose try_send when the producer is on a path that cannot afford to park, and you have an explicit plan for a Full result.

Close-only error model (FS2 parity — no typed fail() in v1)

The terminal model is deliberately close-only. There is one graceful terminal — close() — and no typed fail(error) API in this first version. This mirrors FS2's Channel, which likewise has no typed failure channel. The maintainer's reasoning: keep v1 API-compatible with the primitive it mirrors; runtime failures (a panicking operator, cancellation) still surface through the normal stream error path as Result<T, StreamError>. If a Rust-native fail(StreamError) is ever wanted it can be added later without breaking the close-only contract — adding a terminal is additive; removing one is not.

What this means for you in practice: close() is your only shutdown signal, and it always means "drain what's buffered, then complete successfully." There is no way to make the consumer stream end in an error state from the producer side.

Single active consumer — enforced, not assumed

A channel has room for exactly one active consumer stream. Materializing channel.source() claims that slot with an atomic compare-exchange; a concurrent second materialization fails fast with StreamError::Failed("channel source already has an active consumer") rather than silently splitting the element stream two ways. MPSC means multi-producer, single-consumer — the type enforces the "single" half instead of trusting you to. (If you need many consumers, you want Topic or a BroadcastHub, not a channel.)

Per-producer FIFO; cross-producer order is race-defined

Within a single producer handle, values are delivered in the order that producer sent them — per-producer FIFO. Across producers there is no global ordering guarantee: whichever send wins the race for a ring slot lands first. This is the honest contract of a lock-free MPSC ring, and it is what the benchmarks and tests assert. If you need a total order across producers, impose it yourself (e.g. a sequence number in the payload).

Gentle examples

Every block below is imported from a cargo test–verified integration test.

The smallest channel

Create a bounded channel, materialize its consumer, push a couple of values, and close:

use datum::{Channel, Sink};

// A bounded channel: many producers, exactly one consumer stream.
let channel = Channel::bounded(8);

// The consumer side is an ordinary Source. Materializing it claims the
// single consumer slot and starts draining on the Datum thread pool.
let completion = channel.source().run_with(Sink::collect()).unwrap();

// Producers enqueue without waiting while the ring has room.
channel.try_send("task-1").unwrap();
channel.try_send("task-2").unwrap();

// close() is graceful: buffered items drain, then the source completes.
channel.close();

let items = completion.wait().unwrap();
assert_eq!(items, vec!["task-1", "task-2"]);

Materializing channel.source() is what starts the consumer draining — until then the source is an inert blueprint, exactly like every other Datum Source (blueprint vs. run).

try_send, and awaiting close

try_send never waits: it reports Full when the ring is full right now and Closed after close, returning your value both times. You can also await the close edge instead of polling is_closed():

use datum::{Channel, TrySendError};
use std::thread;

// Capacity 1, and we deliberately do not materialize the consumer yet, so
// nothing drains the ring during this example.
let channel = Channel::<u64>::bounded(1);

// A watcher can await the close edge instead of polling is_closed().
let watcher = {
    let handle = channel.clone();
    thread::spawn(move || {
        futures::executor::block_on(handle.closed());
        "closed"
    })
};

// try_send never waits. The first value fits; the second reports Full and
// hands the rejected value straight back.
channel.try_send(1).unwrap();
assert_eq!(channel.try_send(2), Err(TrySendError::Full(2)));

// After close(), every send path reports Closed and returns the value.
channel.close();
assert_eq!(channel.try_send(3), Err(TrySendError::Closed(3)));

assert_eq!(watcher.join().unwrap(), "closed");

Many producers, backpressured, with a clean shutdown

The realistic shape: several producer threads fan into one consumer using the backpressured send().await, a small capacity forces real waiting, and per-producer order is preserved. This uses the prelude glob import:

use datum::prelude::*;
use std::thread;

// A small capacity forces real backpressure: send() parks (without
// spinning) while the ring is full and resumes when the consumer frees a
// slot. try_send would return Full here instead of waiting.
let channel = Channel::bounded(4);
let completion = channel.source().run_with(Sink::collect()).unwrap();

// Fan three producer threads into the one consumer. Clone the handle per
// producer; the shared channel is the join point.
let mut producers = Vec::new();
for producer_id in 0..3_u64 {
    let handle = channel.clone();
    producers.push(thread::spawn(move || {
        for seq in 0..4_u64 {
            futures::executor::block_on(handle.send((producer_id, seq))).unwrap();
        }
    }));
}
for producer in producers {
    producer.join().unwrap();
}

// Close only after every producer has finished, so nothing is lost.
channel.close();

let delivered = completion.wait().unwrap();
assert_eq!(delivered.len(), 12);

// Cross-producer interleaving is race-defined, but each producer's own
// values always arrive in the order it sent them (per-producer FIFO).
for producer_id in 0..3_u64 {
    let seqs: Vec<u64> = delivered
        .iter()
        .filter(|&&(id, _)| id == producer_id)
        .map(|&(_, seq)| seq)
        .collect();
    assert_eq!(seqs, vec![0, 1, 2, 3]);
}

Channel as an ingestion boundary in front of a pipeline

Because the consumer is a plain Source, you can put processing right behind the channel. Here a callback-style producer (external, non-stream code) forwards events with try_send, and the consumer side filters, maps, and folds them — the channel is the seam between "code that calls in" and "a stream pipeline":

use datum::{Channel, Sink};

// Channel as an ingestion boundary: the consumer Source flows straight into
// filter/map/fold like any other Source.
let channel = Channel::bounded(16);
let completion = channel
    .source()
    .filter(|event| event % 2 == 0)
    .map(|event| event * 10)
    .run_with(Sink::fold(0_u64, |acc, event| acc + event))
    .unwrap();

// Adapt a callback-style producer: some non-stream API hands us events one
// at a time. Forward each into the channel from the callback body.
let emit = {
    let handle = channel.clone();
    move |event: u64| {
        // try_send is the right tool inside a synchronous callback that must
        // not block; branch on Full/Closed to apply your own policy.
        let _ = handle.try_send(event);
    }
};
for event in 0..10_u64 {
    emit(event);
}

// Closing from the original handle drains the buffered events, then completes.
channel.close();

// Even events 0,2,4,6,8 -> *10 -> sum = 200.
assert_eq!(completion.wait().unwrap(), 200);

Using it properly

Correct-usage checklist

• Materialize the consumer. channel.source() is a blueprint; nothing drains until you run it. A common first mistake is holding the source, sending happily until the ring fills, then wondering why send blocks forever (or try_send returns Full) — the buffer is full and no one is reading.
• Close from a producer handle, once producers are done. close() is idempotent and graceful; call it after your producers have finished sending so buffered work still drains.
• Handle the returned value on Full/Closed. Both try_send errors hand the value back. That is your cue to retry, drop, or route it elsewhere — the channel will not hold it for you.
• Size capacity for your handoff, not your peak. The ring is a smoothing buffer, not a backlog store. A larger capacity absorbs more burst before backpressure kicks in but adds latency and memory; a smaller one couples producers tightly to consumer speed. (The consumer drains in batches of up to 256 internally, so very small capacities still handoff efficiently.)

Pitfalls & surprises

• Capacity must be > 0. Channel::bounded(0) (and Source::channel(0)) panics. FS2's synchronous/rendezvous bounded(0) mode is intentionally unsupported in v1.
• Sends after close fail — always. Once closed, both send and try_send return Closed(value). There is no reopening; construct a new channel.
• Double-consume is rejected, not shared. A concurrent second materialization of the same channel source fails with StreamError::Failed. One channel, one live consumer.
• Dropping the consumer stream closes the channel. If the materialized consumer is dropped (e.g. its task ends or is cancelled), the channel closes: blocked producers wake with Closed, and future sends fail. Dropping a producer handle, by contrast, does not close the channel — only the consumer's disappearance does.
• Cross-producer order is not deterministic. Do not assert a global order across producers; only per-producer FIFO holds (see design notes above).
• No typed failure. You cannot fail the consumer from the producer side; the model is close-only.

Integration tips

Channel<T>'s consumer is an ordinary Source, which is what makes it compose everywhere:

• Source / Flow / Sink pipelines. Chain channel.source() into any operator sequence, exactly as the ingestion example shows. It fuses and backpressures like any other source.
• Graphs. Feed channel.source() into a GraphDSL as an inlet source — a good pattern when external events must enter a fan-out/fan-in topology.
• Actors. Bridge a Ractor actor (or any message handler) into a stream: hold a cloned Channel handle in the actor and try_send each message into it, then process the consumer source with ActorFlow::ask or ordinary operators.
• Kill switches. Because the consumer is a normal source, wrap it with a kill switch for external cancellation. Note the interaction: cancelling the stream drops the consumer, which closes the channel and wakes any blocked producers with Closed.
• datum-net. A network read task (a callback or single-owner carrier task) can try_send received frames into a channel, giving downstream a backpressured network ingestion boundary without a bespoke queue.

The Source::channel convenience

Two equivalent construction styles exist. The explicit style is Channel::bounded(capacity) plus channel.source() — you hold the handle first. The convenience style is Source::channel(capacity), which produces a Source<T, Channel<T>> whose materialized value is the channel handle. Reach for the latter when the channel is part of a larger graph and you want the handle to fall out of materialization:

use datum::prelude::*;

let (channel, completion) = Source::<u64>::channel(4)
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

channel.try_send(10).unwrap();
channel.close();
assert_eq!(completion.wait().unwrap(), vec![10]);

Both styles build the same kind of channel. Source::channel creates a fresh channel per materialization and claims its consumer slot immediately.

Imports

The explicit import names exactly what you use; the prelude is equivalent and convenient in Datum-centric modules:

// Explicit
use datum::{Channel, ChannelSendError, Sink, Source, TrySendError};

// Or the prelude glob
use datum::prelude::*;

send's error type is re-exported from the crate root as ChannelSendError (the in-module name is SendError); try_send's is TrySendError.

Performance note

On the channel_mpsc_send_1024xN benchmark (N producers × 1,024 elements → one folding consumer, bounded, backpressured), Datum passes every frozen wall-clock and CPU target against the best JVM competitor per sweep point (ZIO Queue at one producer; Akka MergeHub from 16 producers up). Numbers are from the same-host M9 record captured 2026-07-02:

N producers	Best JVM wall (µs/op)	Datum wall (µs/op)	Best JVM CPU (µs/op)	Datum CPU (µs/op)
1	272 (ZIO Queue)	159	600	250
16	4,979 (Akka MergeHub)	1,496	18,902	3,000
64	18,682 (Akka MergeHub)	4,434	90,538	8,750
256	72,495 (Akka MergeHub)	17,364	355,895	34,167
1,024	293,274 (Akka MergeHub)	38,953	1,435,369	70,000

Overall that is 2.54×–43.44× faster wall-clock and 2.60×–370.23× lower CPU than the best JVM row across the sweep, with allocation far lower too (e.g. ~484 KB/op vs Akka's ~112 MB/op at 1,024 producers). The lever behind the CPU wins is the two-plane design plus batched permit handoff: the consumer drains up to 256 ring elements per poll, releases those slots as one batch, and grants them to parked producers through producer-local permits — rather than waking one task per freed slot.

Full tables (p50/p99, allocation, RSS, correctness counters, and the implementation notes behind each row) are in the concurrency-primitives benchmark record.