Generic subscriber registry for broadcasting to SSE clients.
Supports channel-based filtering — subscribers connect with optional channel filters, and broadcasts reach only matching subscribers. Optional identity keys enable force-closing subscribers by identity (e.g., close all streams for a specific account when their permissions change).
2 declarations
realtime/subscriber_registry.ts view source
Subscriber<T> TstreamSseStream<T>channelsChannels this subscriber listens to. null means all channels.
Set<string> | nullidentityOptional identity key for targeted disconnection (e.g., account_id).
string | nullrealtime/subscriber_registry.ts view source
Generic subscriber registry with channel-based filtering and identity-keyed disconnection.
Subscribers connect with optional channel filters and an optional identity key.
Broadcasts go to a specific channel and reach only matching subscribers.
close_by_identity force-closes all subscribers with a given identity —
use for auth revocation (close streams when a user's permissions change).
Tconst registry = new SubscriberRegistry<SseNotification>();
// subscriber connects (from SSE endpoint)
const unsubscribe = registry.subscribe(stream, ['runs']);
// when a run changes
registry.broadcast('runs', {method: 'run_created', params: {run}});
// subscriber disconnects
unsubscribe();// identity-keyed subscription for auth revocation
const unsubscribe = registry.subscribe(stream, ['audit_log'], account_id);
// when admin revokes the user's role — close their streams
registry.close_by_identity(account_id);subscribeAdd a subscriber.
type (stream: SseStream<T>, channels?: string[] | undefined, identity?: string | undefined): () => void
streamSSE stream to send data to
SseStream<T>channels?channels to subscribe to (undefined or empty = all channels)
string[] | undefinedidentity?optional identity key for targeted disconnection
string | undefined() => voidunsubscribe function
broadcastBroadcast data to all subscribers on a channel.
Subscribers with no channel filter receive all broadcasts. Subscribers with a channel filter only receive matching broadcasts.
type (channel: string, data: T): void
channelthe channel to broadcast on
stringdatathe data to send
Tvoidclose_by_identityForce-close all subscribers with the given identity.
Closes each matching stream and removes the subscriber from the registry. Use for auth revocation — when a user's permissions change, close their SSE connections so they must reconnect and re-authenticate.
type (identity: string): number
identitythe identity key to match
stringnumberthe number of subscribers closed