75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""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")
|