Updated July 1, 2026

On this page

fastware SSE broadcasting: choosing SSE over WebSocket, typed events, per-client async queues, heartbeats, and automatic disconnected-client pruning.

#SSE Broadcasting

Server-Sent Events (SSE) provide a simple, HTTP-based mechanism for pushing data from server to client. Unlike WebSockets, SSE is unidirectional (server to client only), uses plain HTTP, works through proxies and firewalls, and reconnects automatically.

#When to use SSE vs WebSocket

When to use SSE vs WebSocket
Criterion	SSE	WebSocket
Direction	Server to client only	Bidirectional
Protocol	HTTP (text/event-stream)	Upgraded connection (ws://)
Reconnection	Built into the browser (EventSource auto-reconnects)	Manual reconnection logic required
Proxy/firewall	Works through standard HTTP infrastructure	May be blocked by some proxies
Use case	Live dashboards, notifications, progress updates, log tailing	Chat, collaborative editing, gaming

Use SSE when you only need to push data to the client. Use WebSockets when the client needs to send data back over the same connection.

#Basic setup

#1. Create a Broadcaster

python

from fastware import Broadcaster, sse_route

broadcaster = Broadcaster()

The Broadcaster manages a list of connected clients. Each client gets its own async queue. When you broadcast an event, it is pushed to every client's queue.

#2. Register event types

python

broadcaster.register_event("update")
broadcaster.register_event("error")
broadcaster.register_event("heartbeat")

By default, the Broadcaster runs in strict mode -- broadcasting an unregistered event name raises ValueError. This prevents typos and ensures the event vocabulary is explicit.

python

class TestRegisterEvent:
    def test_register_single_event(self):
        b = Broadcaster()
        b.register_event("update")
        assert "update" in b.event_types

    def test_register_multiple_events(self):
        b = Broadcaster()
        b.register_event("update")
        b.register_event("toast")
        assert b.event_types == frozenset({"update", "toast"})

    def test_register_idempotent(self):
        b = Broadcaster()
        b.register_event("update")
        b.register_event("update")
        assert b.event_types == frozenset({"update"})

    def test_event_types_is_frozen(self):
        """The property returns a frozenset -- callers cannot mutate it."""
        b = Broadcaster()
        b.register_event("update")
        with pytest.raises(AttributeError):
            b.event_types.add("sneaky")

python

class TestBroadcastUnregisteredRaises:
    def test_raises_on_unregistered_event(self):
        b = Broadcaster()
        with pytest.raises(ValueError, match="unregistered event type"):
            b.broadcast("unknown", {"key": "val"})

    def test_raises_shows_registered_types(self):
        b = Broadcaster()
        b.register_event("allowed")
        with pytest.raises(ValueError, match="allowed"):
            b.broadcast("nope", {})

    def test_non_strict_skips_validation(self):
        b = Broadcaster(strict=False)
        # Should not raise even with no registered events
        b.broadcast("anything", {"key": "val"})

#3. Wire to a route

python

from fastware import Router, create_app

router = Router()
router.add_route("GET", "/events", sse_route(broadcaster))

The sse_route helper returns an async handler that calls broadcaster.stream(request), which creates a per-client queue and returns a StreamResponse with content-type: text/event-stream.

#4. Broadcast from handlers

python

@router.post("/items")
async def create_item(request):
    item = request.json
    # ... save to database ...
    broadcaster.broadcast("update", {"action": "created", "item": item})
    return {"ok": True}

broadcast() is synchronous -- it pushes the formatted SSE message to every client queue without awaiting. Clients whose queues are full (they fell behind) are pruned automatically.

python

class TestBroadcastDeliversToClient:
    def test_single_client_receives_message(self):
        b = Broadcaster()
        b.register_event("ping")
        q: asyncio.Queue[str] = asyncio.Queue(maxsize=256)
        b._clients.append(q)

        b.broadcast("ping", {"ts": 1})
        assert not q.empty()
        msg = q.get_nowait()
        assert msg.startswith("event: ping\n")
        assert '"ts":1' in msg

    def test_multiple_clients_receive_message(self):
        b = Broadcaster()
        b.register_event("tick")
        queues = [asyncio.Queue(maxsize=256) for _ in range(3)]
        for q in queues:
            b._clients.append(q)

        b.broadcast("tick", {"n": 42})
        for q in queues:
            assert not q.empty()
            msg = q.get_nowait()
            assert "tick" in msg

    def test_string_data_sent_verbatim(self):
        b = Broadcaster()
        b.register_event("raw")
        q: asyncio.Queue[str] = asyncio.Queue(maxsize=256)
        b._clients.append(q)

        b.broadcast("raw", "hello world")
        msg = q.get_nowait()
        assert "data: hello world\n" in msg

python

class TestBroadcastPrunesFullQueue:
    def test_full_queue_is_pruned(self):
        b = Broadcaster(buffer_size=1)
        b.register_event("x")
        q: asyncio.Queue[str] = asyncio.Queue(maxsize=1)
        b._clients.append(q)

        # Fill the queue
        b.broadcast("x", {})
        assert b.client_count == 1

        # Next broadcast overflows -- queue should be pruned
        b.broadcast("x", {})
        assert b.client_count == 0

    def test_healthy_client_survives_alongside_pruned(self):
        b = Broadcaster(buffer_size=1)
        b.register_event("x")
        full_q: asyncio.Queue[str] = asyncio.Queue(maxsize=1)
        healthy_q: asyncio.Queue[str] = asyncio.Queue(maxsize=256)
        b._clients.append(full_q)
        b._clients.append(healthy_q)

        # Fill the small queue
        b.broadcast("x", {"n": 1})
        assert b.client_count == 2

        # Overflow the small queue; healthy one stays
        b.broadcast("x", {"n": 2})
        assert b.client_count == 1
        assert healthy_q in b._clients
        assert full_q not in b._clients

#5. Create the app

python

app = create_app(router)

#Client-side JavaScript

javascript

const source = new EventSource("/events");

source.addEventListener("update", (event) => {
    const data = JSON.parse(event.data);
    console.log("Update received:", data);
});

source.addEventListener("error", (event) => {
    const data = JSON.parse(event.data);
    console.error("Server error:", data);
});

// Connection status
source.onopen = () => console.log("SSE connected");
source.onerror = () => console.log("SSE reconnecting...");

The browser's EventSource automatically reconnects if the connection drops. Events are dispatched by their event: field, which maps to the first argument of broadcaster.broadcast().

python

class TestSSEWireFormat:
    def test_dict_data_is_json_serialized(self):
        b = Broadcaster(strict=False)
        msg = b._format_sse("update", {"key": "value"})
        payload = msgspec.json.encode({"key": "value"}).decode()
        assert msg == f"event: update\ndata: {payload}\n\n"

    def test_string_data_sent_as_is(self):
        b = Broadcaster(strict=False)
        msg = b._format_sse("raw", "plain text")
        assert msg == "event: raw\ndata: plain text\n\n"

    def test_empty_dict(self):
        b = Broadcaster(strict=False)
        msg = b._format_sse("ping", {})
        assert msg == "event: ping\ndata: {}\n\n"

    def test_ends_with_double_newline(self):
        b = Broadcaster(strict=False)
        msg = b._format_sse("ev", {})
        assert msg.endswith("\n\n")

    def test_event_line_comes_first(self):
        b = Broadcaster(strict=False)
        msg = b._format_sse("ev", {"a": 1})
        lines = msg.split("\n")
        assert lines[0] == "event: ev"
        assert lines[1].startswith("data: ")

python

class TestSSERoute:
    def test_returns_callable(self):
        b = Broadcaster()
        handler = sse_route(b)
        assert callable(handler)

#Heartbeat configuration

Long-lived SSE connections can be silently dropped by proxies, load balancers, or firewalls that enforce idle timeouts. Heartbeats prevent this by sending periodic SSE comments (: heartbeat\n\n) that keep the connection alive without triggering client-side event handlers.

python

broadcaster = Broadcaster(heartbeat_interval=30)  # seconds

When heartbeat_interval is set, the event generator sends a comment line if no real event arrives within the interval. SSE comments (lines starting with :) are ignored by EventSource -- they keep the TCP connection alive without producing a JavaScript event.

If heartbeat_interval is None (the default), no heartbeats are sent and the generator blocks indefinitely waiting for real events.

#Strict mode vs permissive mode

The Broadcaster supports 2 event validation modes that control whether event type names must be pre-registered before broadcasting. Strict mode (the default) catches typos at development time; permissive mode allows dynamic event vocabularies:

Strict mode (default, strict=True):

python

broadcaster = Broadcaster(strict=True)
broadcaster.register_event("update")
broadcaster.broadcast("update", {"ok": True})   # works
broadcaster.broadcast("typo", {"ok": True})     # raises ValueError

Strict mode catches typos and enforces a declared event vocabulary. Register all event types before broadcasting.

Permissive mode (strict=False):

python

broadcaster = Broadcaster(strict=False)
broadcaster.broadcast("anything", {"ok": True})  # works without registration

Permissive mode skips event type validation. Use this when event types are dynamic or user-defined.

#Buffer size

Each client gets an async queue with a configurable maximum size. The default buffer holds 256 messages per client. Slow consumers whose queues fill up are automatically pruned on the next broadcast() call to prevent unbounded memory growth:

python

broadcaster = Broadcaster(buffer_size=512)

When a client's queue is full (the client is not consuming messages fast enough), the client is pruned from the client list on the next broadcast() call. This prevents a slow consumer from causing memory growth.

python

class TestClientCount:
    def test_initially_zero(self):
        b = Broadcaster()
        assert b.client_count == 0

    def test_tracks_appended_clients(self):
        b = Broadcaster()
        b._clients.append(asyncio.Queue(maxsize=256))
        assert b.client_count == 1
        b._clients.append(asyncio.Queue(maxsize=256))
        assert b.client_count == 2

#Introspection

python

broadcaster.client_count   # number of connected clients
broadcaster.event_types    # frozenset of registered event types

#Complete example

python

from fastware import Router, Broadcaster, sse_route, create_app, serve

broadcaster = Broadcaster(heartbeat_interval=30)
broadcaster.register_event("message")
broadcaster.register_event("status")

router = Router()
router.add_route("GET", "/events", sse_route(broadcaster))

@router.post("/send")
async def send_message(request):
    data = request.json
    broadcaster.broadcast("message", {"text": data["text"]})
    return {"sent": True}

@router.get("/status")
async def get_status(request):
    broadcaster.broadcast("status", {"clients": broadcaster.client_count})
    return {"clients": broadcaster.client_count}

app = create_app(router)

if __name__ == "__main__":
    serve(app, foreground=True, host="127.0.0.1", port=8000)

#API reference

See the full Broadcaster class reference below, which documents the register_event method for declaring event types, the broadcast method for pushing events to all connected clients, the stream method for creating per-client SSE response generators, and the sse_route helper function for wiring a Broadcaster to a route:

#`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`

python

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

Declare an allowed event type.

#`event_types`

python

def event_types(self) -> frozenset[str]

Currently registered event types.

#`_format_sse`

python

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

Format a payload as an SSE wire message.

Dict payloads are serialized with msgspec (project convention). A multi-line payload is emitted as one data: line per line, per the SSE spec, so a stray newline in the payload cannot terminate the event early or inject additional SSE fields.

#`broadcast`

python

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`

python

async def _event_generator(self, queue: asyncio.Queue[str], initial: list[tuple[str, dict[str, Any] | str]] | None=None) -> AsyncGenerator[str, None]

Yield SSE messages from a per-client queue.

The queue is registered as a client only once iteration begins, and the finally block guarantees it is unregistered when the generator is closed (e.g. client disconnect). Registering here — rather than in stream() — ensures a StreamResponse whose body is never consumed does not leak a queue into self._clients.

initial events are formatted and yielded once, before the queue loop, so a connection can be primed with current state (e.g. the current build id on the update channel). They bypass the strict registration check -- the caller controls them, not a broadcast.

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

#`stream`

python

async def stream(self, request: Request, initial: list[tuple[str, dict[str, Any] | str]] | None=None) -> StreamResponse

Return a StreamResponse for an SSE endpoint.

Creates a per-client queue and wraps the async generator in the framework's streaming response type. The queue is registered as a client by _event_generator when iteration starts, not here, so an unconsumed response never leaks a queue.

initial is an optional list of (event, data) pairs sent to this connection before any broadcast, so a client can be primed with the current state on connect.

#`client_count`

python

def client_count(self) -> int

Number of currently connected SSE clients.

#`sse_route`

python

def sse_route(broadcaster: Broadcaster)

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

#SSE Broadcasting

#When to use SSE vs WebSocket

#Basic setup

#1. Create a Broadcaster

#2. Register event types

#3. Wire to a route

#4. Broadcast from handlers

#5. Create the app

#Client-side JavaScript

#Heartbeat configuration

#Strict mode vs permissive mode

#Buffer size

#Introspection

#Complete example

#API reference

#src.fastware.sse

#Broadcaster

#register_event

#event_types

#_format_sse

#broadcast

#_event_generator

#stream

#client_count

#sse_route

#`src.fastware.sse`

#`Broadcaster`

#`register_event`

#`event_types`

#`_format_sse`

#`broadcast`

#`_event_generator`

#`stream`

#`client_count`

#`sse_route`