Historical post from the 3.x line. The snippet below uses removed APIs (app.mount, bootstrap without yield). Current patterns: Getting started and Runtime.
Stario 3.2.0: connection lifecycle on Context (c.closing, c.alive()), app.shutting_down for tasks with no active request, and the single Writer per response. Datastar default + vendored bundle at v1.0.1.
Work queue
POST validates and Relay.publish commands; one long-lived GET owns the only Writer that writes that stream's body; a worker started with app.create_task subscribes and republishes (or drains a queue only that GET reads)—one coroutine touches Writer for that socket. Stream side: c.closing / c.alive(). Worker side: app.shutting_down; idle relays block on __anext__, so use asyncio.wait vs shutdown when you need SIGINT alone to unblock.
Declare the worker beside bootstrap, schedule once, pass relay into feature registration like chat-room (register_lobby / register_room). Async-generator bootstrap with yield gives a teardown slice after shutdown starts (Runtime — lifecycle).
from collections.abc import AsyncIterator from stario import App, Relay, Span async def command_bridge(app: App, relay: Relay[str]) -> None: # async for on subscribe() registers for the whole loop (same as async with + for). async for topic, data in relay.subscribe("cmd.*"): relay.publish(f"stream:{topic}", f"applied:{data}") if app.shutting_down: break async def bootstrap(app: App, span: Span) -> AsyncIterator[None]: relay: Relay[str] = Relay() span.attr("relay", "control-plane") app.create_task(command_bridge(app, relay), name="cmd-bridge") app.mount("/", build_router(relay)) yield— Adam