brokkr-broker::ws::broadcaster

Rust

In-memory per-stack fan-out of agent telemetry frames.

Sits between the WS reader (dispatch_uplink) and the live-tail subscription endpoint (WS-11). Agents stream K8sEvent / PodLogLine / LogGap frames upstream; this hub broadcasts them, unmodified, to any UI/SDK clients currently subscribed to the originating stack.

Slow-subscriber policy

Each per-stack channel is bounded (CHANNEL_CAPACITY). Subscribers that fall behind get RecvError::Lagged(n) from tokio::sync::broadcast, which the subscription handler converts into a synthetic LogGap so the UI can render a visible gap rather than silently dropping frames. This is per ADR-0008: “a slow subscriber must not slow ingestion”.

Structs

struct LiveBroadcaster

pub

Derives: Default

Methods

new

pub

fn new() -> Arc<Self>

Source

    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

broadcast

pub

fn broadcast(&self, stack_id: Uuid, msg: WsMessage)

Send a frame to every subscriber of stack_id. No-op when nobody is subscribed; broadcast::Sender::send errors are intentionally swallowed (an empty channel returns SendError).

Source

    pub fn broadcast(&self, stack_id: Uuid, msg: WsMessage) {
        let channels = self.channels.read().expect("broadcaster poisoned");
        if let Some(tx) = channels.get(&stack_id) {
            let _ = tx.send(msg);
        }
    }

subscribe

pub

fn subscribe(&self, stack_id: Uuid) -> broadcast::Receiver<WsMessage>

Subscribe to all future frames for stack_id. Creates the underlying channel lazily on first subscription.

Source

    pub fn subscribe(&self, stack_id: Uuid) -> broadcast::Receiver<WsMessage> {
        let mut channels = self.channels.write().expect("broadcaster poisoned");
        let sender = channels
            .entry(stack_id)
            .or_insert_with(|| broadcast::channel(CHANNEL_CAPACITY).0);
        sender.subscribe()
    }

stack_​count

pub

fn stack_count(&self) -> usize

Diagnostics: number of stacks with at least one live subscriber.

Source

    pub fn stack_count(&self) -> usize {
        self.channels.read().expect("broadcaster poisoned").len()
    }

subscriber_​count

pub

fn subscriber_count(&self) -> usize

Diagnostics: total subscriber count across all stacks.

Source

    pub fn subscriber_count(&self) -> usize {
        self.channels
            .read()
            .expect("broadcaster poisoned")
            .values()
            .map(|tx| tx.receiver_count())
            .sum()
    }

struct FleetBroadcaster

pub

In-memory fleet-wide fan-out of per-agent FleetUpdate frames (BROKKR-I-0028).

Unlike LiveBroadcaster, this is a single channel, not keyed per stack: every /api/v1/fleet/live subscriber receives every agent’s updates. Same slow-subscriber policy as the per-stack hub (ADR-0008): a lagging subscriber gets RecvError::Lagged and the handler drops/continues — broadcasting never blocks the triggering operation.

Methods

new

pub

fn new() -> Arc<Self>

Source

    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

broadcast

pub

fn broadcast(&self, msg: WsMessage)

Broadcast one frame to every fleet subscriber. No-op when nobody is subscribed (broadcast::Sender::send returns SendError, swallowed). Never blocks — this is called from producer hot paths where a push failure must never affect the triggering operation.

Source

    pub fn broadcast(&self, msg: WsMessage) {
        let _ = self.tx.send(msg);
    }

subscribe

pub

fn subscribe(&self) -> broadcast::Receiver<WsMessage>

Subscribe to all future fleet frames.

Source

    pub fn subscribe(&self) -> broadcast::Receiver<WsMessage> {
        self.tx.subscribe()
    }

subscriber_​count

pub

fn subscriber_count(&self) -> usize

Diagnostics: current number of fleet-live subscribers.

Source

    pub fn subscriber_count(&self) -> usize {
        self.tx.receiver_count()
    }