BroadcastChannel
@rotorsoft/act-root / act-http/src/sse / BroadcastChannel
Class: BroadcastChannel<S>
Defined in: libs/act-http/src/sse/broadcast.ts:149
Server-side broadcast channel for incremental state sync over SSE.
Manages per-stream subscriber sets and an LRU state cache. When state changes, forwards domain patches (from event handlers) to all subscribers as version-keyed messages.
Usage
const broadcast = new BroadcastChannel<MyState>();
// After every app.do(), publish the state and its patches (how to
// build both, a `.snap`'s empty patch included:
// https://rotorsoft.github.io/act-root/docs/concepts/real-time#broadcasting-a-commit):
broadcast.publish(streamId, state, patches);
// In SSE subscription:
const cleanup = broadcast.subscribe(streamId, (msg) => {
pending = msg;
resolve?.();
});
// Initial state for reconnects:
const cached = broadcast.state(streamId);
Version Contract
The _v field on state MUST be set from snap.version (the event store's
monotonic stream version at the head) BEFORE calling publish(). This is
the single source of truth for ordering — no separate version counters.
When .snap fired, the snapshot holds that version: pass an empty patch for
it, so the patches stay contiguous with the next commit.
Type Parameters
S
S extends BroadcastState = BroadcastState
Constructors
Constructor
new BroadcastChannel<
S>(options?):BroadcastChannel<S>
Defined in: libs/act-http/src/sse/broadcast.ts:155
Parameters
options?
cacheSize?
number
Max number of stream states kept in the LRU cache (default 50).
onOverlayMiss?
(streamId) => void
Called when overlay-contributed state is lost, on either of the two
paths that can lose it: overlay() finding no cached baseline, and
the LRU evicting an entry that carried overlay keys — the latter from
inside publish(), for a stream other than the one being published.
Live subscribers of the affected stream are sent a resync frame, so
they refetch rather than keep showing state the server has forgotten.
The hook is observability on top of that recovery: a steady stream of
calls means cacheSize is too small for the working set.
A throw is caught and routed to onSubscriberError — the resync has
already gone out, and a host bug must not cost the publishing stream
its own frame. Defaults to a no-op.
onSubscriberError?
(error, streamId) => void
Called when a subscriber callback throws. The frame is still delivered
to every other subscriber and the publish still returns normally — a
bad consumer must not break the publisher. Defaults to console.error;
pass your logger here to route it with the rest of your logs.
Returns
BroadcastChannel<S>
Accessors
cache
Get Signature
get cache():
StateCache<S>
Defined in: libs/act-http/src/sse/broadcast.ts:345
Direct access to the state cache (for app-specific reads like presence).
Returns
StateCache<S>
Methods
overlay()
overlay(
streamId,overlay_patch):PatchMessage<S> |undefined
Defined in: libs/act-http/src/sse/broadcast.ts:274
Publish a state update that doesn't change the event version (e.g. presence overlay, computed field refresh). Uses the same version as the cached state, single entry.
Parameters
streamId
string
overlay_patch
Partial<S>
Returns
PatchMessage<S> | undefined
publish()
publish(
streamId,state,patches?):PatchMessage<S>
Defined in: libs/act-http/src/sse/broadcast.ts:229
Publish domain patches from a commit. patches[i] corresponds to version baseV + i + 1.
Parameters
streamId
string
The event store stream ID
state
S
Full state with _v set from snap.version
patches?
Readonly<Patch<S>>[] = []
Array of domain patches, one per emitted event, as the
snapshots carry them (null deletes a key)
Returns
PatchMessage<S>
resync()
resync(
streamId):void
Defined in: libs/act-http/src/sse/broadcast.ts:334
Tells a stream's clients to refetch, and forgets its cached state. For a
stream whose versions started over: after close({ restart: true }), a
client ahead of the new versions would read every patch as stale.
Parameters
streamId
string
Returns
void
state()
state(
streamId):S|undefined
Defined in: libs/act-http/src/sse/broadcast.ts:340
Get the cached state for a stream (for reconnects / initial SSE yield).
Parameters
streamId
string
Returns
S | undefined
subscribe()
subscribe(
streamId,cb): () =>void
Defined in: libs/act-http/src/sse/broadcast.ts:313
Subscribe to broadcast messages for a stream. Returns a cleanup function that removes the subscription.
Parameters
streamId
string
cb
Subscriber<S>
Returns
() => void
subscriberCount()
subscriberCount(
streamId):number
Defined in: libs/act-http/src/sse/broadcast.ts:325
Get the number of subscribers for a stream.
Parameters
streamId
string
Returns
number