update

backend/app/events.py

@@ -0,0 +1,62 @@
+"""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()
+
+
+def anmelden():
+    queue = asyncio.Queue(maxsize=PUFFER)
+    _abonnenten.add(queue)
+    return queue
+
+
+def abmelden(queue):
+    _abonnenten.discard(queue)
+
+
+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)