Struct Broker 

pub struct Broker {
    inner: Mutex<BrokerInner>,
    observers: RwLock<Vec<Arc<dyn Fn(&MuxEvent) + Send + Sync>>>,
}

The central pub/sub broker for MuxEvents.

Fields

inner: Mutex<BrokerInner>

observers: RwLock<Vec<Arc<dyn Fn(&MuxEvent) + Send + Sync>>>

In-process observers, called on every publish BEFORE the inner lock is taken. They must be cheap and must not publish synchronously — the notification Router’s observer only parses and enqueues.

Implementations

impl Broker

pub fn new() -> Self

pub fn add_observer(&self, f: Arc<dyn Fn(&MuxEvent) + Send + Sync>)

Observe every published event in-process (the broker otherwise has a single fan-out client — the WS bridge). Used by the notification Router to watch agentfailure without touching its publish sites.

pub fn set_client(&self, client: Box<dyn WpsClient>)

pub fn subscribe(&self, route_id: &str, sub: SubscriptionRequest)

Subscribe a route to an event, optionally scoped.

Replay-on-subscribe: after registering the route, immediately deliver any persisted events that match the subscription. Lets late subscribers catch up on the most recent state without waiting for the next publish — closes the race for live-log streaming where the frontend learns the tool_use_id only after the wrapper has already finished publishing.

fn replay_to_route( inner: &mut BrokerInner, route_id: &str, sub: &SubscriptionRequest, )

Deliver any persisted events matching sub to route_id. Called inside subscribe so replay happens atomically under the broker lock — no live event published mid-replay can interleave.

Once-per-(route, event, scope). The frontend (frontend/app/store/mps.ts) flushes eventsub on every listener add/remove; each WebSocket connection has its own conn_id as the route key (PR #1418). Replaying persisted history on each of those flushes would re-emit completed bash logs every pane mount / tab switch / sibling subscription. The replayed set tracks tuples that already received their backfill and short-circuits subsequent resubscribes. Cleared per-route in unsubscribe_all so a true reconnect (route dropped + re-registered) gets a fresh replay.

Star-scope replay is intentionally not implemented (rare, requires scanning every persist key; can add later if needed).

pub fn has_subscribers(&self, event_name: &str) -> bool

Whether any route is subscribed to event_name, in any scope. For a publisher that can skip measuring what nobody reads.

pub fn unsubscribe(&self, route_id: &str, event_name: &str)

Unsubscribe a route from a specific event.

pub fn unsubscribe_all(&self, route_id: &str)

Unsubscribe a route from all events.

Also clears the replayed tracker for this route so a future reconnect (route registers again from scratch) gets a fresh replay of persisted history. Without this, a transient WebSocket drop would silently lose all subsequent replay.

pub fn purge_scope(&self, scope: &str)

Purge all persisted history for a scope (e.g. block:<id> or shell:<id>), regardless of event name. Call when the underlying block/shell is deleted — persist_map’s key set is otherwise never pruned (each key’s event Vec is capped, but the map only grows over a session’s cumulative block/shell count). Also drops matching replayed entries so a stale scope can’t linger across route reconnects.

fn unsubscribe_nolock(inner: &mut BrokerInner, route_id: &str, event_name: &str)

pub fn publish_event<E: Event>( &self, payload: &E::Payload, scopes: Vec<String>, persist: usize, )

Publish a typed event (mps_events): payload is E::Payload, so its frontend subscribers get it as that type.

pub fn publish_event_for<E: PerIdEvent>( &self, id: &str, payload: &E::Payload, scopes: Vec<String>, persist: usize, )

Publishes a per-id event (mps_events::PerIdEvent) under its prefix plus id.

pub fn publish(&self, event: MuxEvent)

Publish an event to all matching subscribers.

pub fn read_event_history( &self, event_type: &str, scope: &str, max_items: usize, ) -> Vec<MuxEvent>

Read persisted event history.

fn persist_event(inner: &mut BrokerInner, event: &MuxEvent)

fn get_matching_routes(inner: &BrokerInner, event: &MuxEvent) -> Vec<String>

Trait Implementations

impl Default for Broker

fn default() -> Self

Returns the “default value” for a type.