#src.fastware.sse

#src.fastware.sse

SSE (Server-Sent Events) broadcaster with typed event registration, per-client async queues, automatic disconnect pruning, and strict mode enforcement.

#Broadcaster

Manages SSE client connections and broadcasts typed events.

Event types must be registered via register_event before they can be broadcast. In strict mode (the default), broadcasting an unregistered event raises ValueError. Pass strict=False to skip validation.

#register_event

def register_event(self, name: str) -> None

Declare an allowed event type.

#event_types

def event_types(self) -> frozenset[str]

Currently registered event types.

#_format_sse

def _format_sse(self, event: str, data: dict[str, Any] | str) -> str

Format a payload as an SSE wire message.

#broadcast

def broadcast(self, event: str, data: dict[str, Any] | str) -> None

Send an event to all connected clients.

Prunes clients whose queues are full (they fell behind and are presumed disconnected or stuck).

Raises ValueError if event was not previously registered and the broadcaster is in strict mode.

#_event_generator

async def _event_generator(self, queue: asyncio.Queue[str]) -> AsyncGenerator[str, None]

Yield SSE messages from a per-client queue.

When heartbeat_interval is set, yields SSE comment heartbeats (": heartbeat\n\n") if no real message arrives within the interval.

#stream

async def stream(self, request: Request) -> StreamResponse

Return a StreamResponse for an SSE endpoint.

Creates a per-client queue, registers it, and wraps the async generator in the framework's streaming response type.

#client_count

def client_count(self) -> int

Number of currently connected SSE clients.

#sse_route

def sse_route(broadcaster: Broadcaster)

Return an async handler suitable for router.add_route("GET", "/events", handler).