"""Live-Kanal zu allen offenen Seiten. Server-Sent Events reichen: Es fließt nur in eine Richtung, und der Browser verbindet nach einem Abbruch von selbst neu. Ein WebSocket wäre mehr Technik für dasselbe Ergebnis. Die Abonnenten liegen im Speicher. Das trägt genau so weit wie ein einzelner Prozess — bei mehreren uvicorn-Workern bekäme jeder nur seine eigenen. """ import asyncio import json import logging log = logging.getLogger("events") # Wenn ein Browser nicht mitkommt, sollen ihn alte Ereignisse nicht # unbegrenzt im Speicher stapeln PUFFER = 16 _abonnenten: set[asyncio.Queue] = set() # Welche Sitzung hinter einer Verbindung steckt — daraus ergibt sich, # wer gerade online ist: eine offene Seite mit gewähltem Namen _sitzungen: dict[asyncio.Queue, str] = {} def anmelden(session_id=None): queue = asyncio.Queue(maxsize=PUFFER) _abonnenten.add(queue) if session_id: _sitzungen[queue] = session_id return queue def abmelden(queue): _abonnenten.discard(queue) _sitzungen.pop(queue, None) def sitzungen_online(): return set(_sitzungen.values()) def anzahl(): return len(_abonnenten) def senden(typ, daten=None): """An alle offenen Seiten schicken. Blockiert nie.""" nachricht = json.dumps({"type": typ, "data": daten}, ensure_ascii=False) for queue in list(_abonnenten): try: queue.put_nowait(nachricht) except asyncio.QueueFull: log.warning("Abonnent kommt nicht mit, Ereignis verworfen") async def stream(queue, request): """SSE-Datenstrom. Der Kommentar alle 20 s hält Proxys davon ab, die Verbindung als tot zu schließen.""" try: yield ": verbunden\n\n" while True: if await request.is_disconnected(): break try: nachricht = await asyncio.wait_for(queue.get(), timeout=20) yield f"data: {nachricht}\n\n" except asyncio.TimeoutError: yield ": ping\n\n" finally: abmelden(queue) # Die anderen Seiten sollen den grünen Punkt ausmachen senden("users")