Struct BroadcastChannel 

pub struct BroadcastChannel<T> { /* private fields */ }

Broadcast channel for multi-consumer scenarios.

Uses a shared ring buffer with per-subscriber cursors for memory efficiency. Slowest consumer determines retention. Supports dynamic subscribe/unsubscribe.

Lock-Free Hot Path

The hot path methods (broadcast, read, slowest_cursor) are completely lock-free, using only atomic operations on the pre-allocated cursor slots. This achieves consistent sub-100ns latency without the 5-50μs spikes that RwLock can cause.

Type Parameters

• T - The event type. Must be Clone for subscribers (typically Arc<RecordBatch> where clone is an O(1) atomic increment).

Performance Targets

Operation	Target
broadcast() (2 subs)	< 100ns
broadcast() (4 subs)	< 150ns
read()	< 50ns

Example

use laminar_core::streaming::broadcast::{BroadcastChannel, BroadcastConfig};

let channel = BroadcastChannel::<i32>::new(BroadcastConfig::default());

// Subscribe
let sub1 = channel.subscribe("mv1").unwrap();
let sub2 = channel.subscribe("mv2").unwrap();

// Broadcast
channel.broadcast(42).unwrap();

// Each subscriber receives the value
assert_eq!(channel.read(sub1), Some(42));
assert_eq!(channel.read(sub2), Some(42));

Implementations

impl<T> BroadcastChannel<T>

pub fn new(config: BroadcastConfig) -> Self

Creates a new broadcast channel with the given configuration.

Pre-allocates all cursor slots for lock-free hot path operation.

Arguments

• config - Channel configuration

Example

let channel = BroadcastChannel::<i32>::new(BroadcastConfig::with_capacity(256));

pub fn slowest_cursor(&self) -> u64

Returns the slowest cursor position among active subscribers.

Lock-free: Only uses atomic loads, no locks.

Returns u64::MAX if there are no active subscribers.

pub fn subscriber_lag(&self, subscriber_id: usize) -> u64

Returns the lag (unread messages) for a subscriber.

Lock-free: Only uses atomic loads, no locks.

Returns 0 if the subscriber ID is out of bounds or inactive.

pub fn subscriber_count(&self) -> usize

Returns the number of active subscribers.

Lock-free: Returns the atomically-maintained count.

pub fn is_lagging(&self, subscriber_id: usize) -> bool

Returns true if the subscriber is lagging beyond the warning threshold.

Lock-free: Only uses atomic loads.

pub fn write_position(&self) -> u64

Returns the current write position.

pub fn capacity(&self) -> usize

Returns the buffer capacity.

pub fn config(&self) -> &BroadcastConfig

Returns the configuration.

pub fn is_closed(&self) -> bool

Returns true if the channel is closed.

pub fn close(&self)

Closes the channel.

After closing, broadcast() returns Err(Closed) and subscribers can only read remaining buffered data.

pub fn subscriber_info(&self, subscriber_id: usize) -> Option<SubscriberInfo>

Returns subscriber information for debugging.

Note

Takes a read lock to access the name. Not intended for hot path use.

Panics

Panics if the internal lock is poisoned (should not happen in normal use).

pub fn list_subscribers(&self) -> Vec<SubscriberInfo>

Lists all active subscribers.

Note

Takes a read lock to access names. Not intended for hot path use.

Panics

Panics if the internal lock is poisoned (should not happen in normal use).

pub fn unsubscribe(&self, subscriber_id: usize)

Unsubscribes a subscriber.

Lock-free: Only uses atomic store on the slot.

The subscriber’s cursor is deactivated but the slot remains allocated (can be reused by future subscribers). Subsequent reads with this ID will return None.

impl<T: Clone> BroadcastChannel<T>

pub fn broadcast(&self, value: T) -> Result<(), BroadcastError>

Broadcasts a value to all subscribers.

Lock-free: Only uses atomic operations, no locks on hot path.

Writes the value into the next available slot. All active subscribers will be able to read this value via read().

Errors

• BroadcastError::NoSubscribers if there are no active subscribers
• BroadcastError::SlowSubscriberTimeout if Block policy times out
• BroadcastError::Closed if the channel is closed

Safety Contract

Must be called from a single writer thread only. The DAG executor enforces this by assigning exactly one producer per broadcast channel.

pub fn subscribe( &self, name: impl Into<String>, ) -> Result<usize, BroadcastError>

Registers a new subscriber.

Returns the subscriber ID which can be used with read(). The subscriber ID is the slot index for O(1) direct access on the hot path.

New subscribers start reading from the current write position (they don’t see historical data).

Errors

Returns BroadcastError::MaxSubscribersReached if all slots are occupied.

Performance

Uses CAS to claim a slot, then takes write lock for name storage. Should only be called during setup, not on hot path.

Panics

Panics if the internal lock is poisoned (should not happen in normal use).

pub fn read(&self, subscriber_id: usize) -> Option<T>

Reads the next value for a subscriber.

Lock-free: Uses direct O(1) array indexing, no locks.

Returns Some(value) if data is available, or None if the subscriber is caught up with the producer, has been unsubscribed, or the ID is invalid.

Arguments

• subscriber_id - The subscriber’s ID from subscribe()

pub fn try_read( &self, subscriber_id: usize, ) -> Result<Option<T>, BroadcastError>

Tries to read without blocking.

Lock-free: Inlined logic, no double-locking.

Returns Ok(Some(value)) if data is available, Ok(None) if caught up, or Err if the subscriber is invalid or unsubscribed.

Errors

Returns BroadcastError::SubscriberNotFound if the subscriber ID is invalid or the subscriber has been unsubscribed.

Trait Implementations

impl<T> Debug for BroadcastChannel<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T: Send> Send for BroadcastChannel<T>

impl<T: Send> Sync for BroadcastChannel<T>