55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
"""Live-Board-Hub: Statuswechsel aus der DB-Schicht → alle WebSocket-Clients.
|
|
Kein Polling; das Frontend hält den Snapshot und wendet Deltas an."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
log = logging.getLogger("creator2.ws")
|
|
|
|
|
|
class Hub:
|
|
def __init__(self):
|
|
self._clients: set = set()
|
|
self._loop: asyncio.AbstractEventLoop | None = None
|
|
self._queue: asyncio.Queue | None = None
|
|
|
|
def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None:
|
|
self._loop = loop
|
|
self._queue = asyncio.Queue()
|
|
# EIN Sender-Task serialisiert die Broadcasts — ein Task-pro-push konnte
|
|
# sich bei Backpressure überholen (out-of-order-Deltas, konkurrierende
|
|
# send_text auf demselben Socket).
|
|
loop.create_task(self._sende_schleife())
|
|
|
|
async def _sende_schleife(self) -> None:
|
|
assert self._queue is not None
|
|
while True:
|
|
msg = await self._queue.get()
|
|
await self._send_all(msg)
|
|
|
|
async def connect(self, websocket) -> None:
|
|
await websocket.accept()
|
|
self._clients.add(websocket)
|
|
|
|
def disconnect(self, websocket) -> None:
|
|
self._clients.discard(websocket)
|
|
|
|
def push(self, tabelle: str, row: dict) -> None:
|
|
"""Thread-sicher (DB-Schicht ruft synchron): in die Queue legen, der
|
|
Sender-Task verschickt der Reihe nach."""
|
|
if not self._clients or self._loop is None or self._queue is None:
|
|
return
|
|
msg = json.dumps({"tabelle": tabelle, "row": row}, ensure_ascii=False, default=str)
|
|
self._loop.call_soon_threadsafe(self._queue.put_nowait, msg)
|
|
|
|
async def _send_all(self, msg: str) -> None:
|
|
for ws in list(self._clients):
|
|
try:
|
|
await ws.send_text(msg)
|
|
except Exception:
|
|
self._clients.discard(ws)
|
|
|
|
|
|
hub = Hub()
|