In-process fan-out message bus for SSE subscribers.
| 24 | |
| 25 | |
| 26 | class _EventBus: |
| 27 | """In-process fan-out message bus for SSE subscribers.""" |
| 28 | |
| 29 | def __init__(self) -> None: |
| 30 | self._lock = threading.Lock() |
| 31 | self._subscribers: list[queue.Queue] = [] |
| 32 | |
| 33 | def subscribe(self) -> queue.Queue: |
| 34 | """Register a new subscriber and return its dedicated queue.""" |
| 35 | q: queue.Queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) |
| 36 | with self._lock: |
| 37 | self._subscribers.append(q) |
| 38 | return q |
| 39 | |
| 40 | def unsubscribe(self, q: queue.Queue) -> None: |
| 41 | """Remove a subscriber queue (called when the HTTP connection closes).""" |
| 42 | with self._lock: |
| 43 | try: |
| 44 | self._subscribers.remove(q) |
| 45 | except ValueError: |
| 46 | pass |
| 47 | |
| 48 | def publish(self, event_type: str, data: Any) -> None: |
| 49 | """ |
| 50 | Publish an event to all active subscribers. |
| 51 | |
| 52 | ``event_type`` should be one of ``"snapshot"`` or ``"event"``. |
| 53 | ``data`` must be JSON-serialisable. |
| 54 | """ |
| 55 | payload = json.dumps(data, ensure_ascii=False) |
| 56 | msg = {"event_type": event_type, "payload": payload} |
| 57 | with self._lock: |
| 58 | dead: list[queue.Queue] = [] |
| 59 | for q in self._subscribers: |
| 60 | try: |
| 61 | q.put_nowait(msg) |
| 62 | except queue.Full: |
| 63 | # Subscriber is too slow: drop the oldest item and retry |
| 64 | try: |
| 65 | q.get_nowait() |
| 66 | q.put_nowait(msg) |
| 67 | except (queue.Empty, queue.Full): |
| 68 | dead.append(q) |
| 69 | for q in dead: |
| 70 | try: |
| 71 | self._subscribers.remove(q) |
| 72 | except ValueError: |
| 73 | pass |
| 74 | |
| 75 | |
| 76 | # Module-level singleton – imported by api.py and core.py |