Kitto.Notification.EventBus

In-process publish/subscribe event bus for the KittoX notification subsystem. Publishers push a TKXEvent onto a topic; each subscriber owns a TKXEventQueue (typically one per open SSE connection) that the bus fills. The bus is thread-safe and process-global (single instance); it is deliberately in-process only — multi-instance fan-out (Redis/NATS) is a future strategy behind the same API.

TKXEventQueue class

A bounded, thread-safe FIFO of events for a single subscriber. A dedicated waiter (the SSE handler thread) blocks on Dequeue until an event arrives or the timeout elapses; publishers call Enqueue from any thread. When the queue is full the oldest event is dropped (bounded backpressure), so a slow/stuck consumer can never exhaust memory.

constructor Create(const AMaxSize: Integer = 1000);

Creates the queue with the given maximum retained events.

procedure Enqueue(const AEvent: TKXEvent);

Appends an event, dropping the oldest if the queue is full.

function Dequeue(out AEvent: TKXEvent;

Waits up to ATimeoutMs for an event. Returns True and the event when one is available, or False on timeout (used by the SSE loop to emit a keep-alive).

procedure Clear;

Discards all pending events.

TKXEventBus class

Process-global publish/subscribe bus. Maps each topic to the set of subscriber queues and delivers a published event to every queue on its topic. The bus owns the per-topic subscriber lists but never the queues themselves (each queue is owned and freed by its subscriber).

procedure Subscribe(const ATopic: string;

Adds a queue as a subscriber of the given topic.

procedure Unsubscribe(const ATopic: string;

Removes a queue from the given topic.

procedure UnsubscribeAll(const AQueue: TKXEventQueue);

Removes a queue from every topic it is subscribed to (call on connection close, before freeing the queue).

procedure Publish(const AEvent: TKXEvent);

Delivers the event to all queues subscribed to its Topic.

procedure PublishToUser(const AUser: string;

Delivers the event to the given user's personal topic (user:{AUser}), overriding the event's Topic.

procedure PublishSystem(const AEvent: TKXEvent);

Delivers the event to the system-wide topic.

class function Instance: TKXEventBus;

The process-global event bus instance.