"""In-process WebSocket fan-out hub.

The poller (one task) writes new event rows here. Every connected

WebSocket (zero or many tasks) reads them. No persistence, no replay

beyond what's already on the secondary — if a client reconnects, it

gets future events but not past ones. Forks that need replay should

have the WebSocket handler do an initial `SELECT … WHERE id <= cursor`

on connect before subscribing.

Why a Hub instead of a single asyncio.Queue per client:

A single queue can only have one consumer (each .get() pops). We

want every connected browser to see every event, so we maintain a

set of per-subscriber queues and the publisher writes to all of

them.

"""

from __future__ import annotations

import asyncio

import logging

from collections.abc import AsyncIterator

from contextlib import asynccontextmanager

from typing import Any

log = logging.getLogger(__name__)

class Hub:

"""Fan-out broadcaster. One publisher, N subscribers, in-memory only."""

def __init__(self, queue_max: int = 1024) -> None:

# Each subscribing WebSocket gets its own bounded queue. Bound is

# high enough to absorb a brief stall but low enough that a

# disconnected/slow client doesn't pin unbounded RAM.

self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()

self._lock = asyncio.Lock()

self._queue_max = queue_max

async def publish(self, event: dict[str, Any]) -> None:

"""Push an event to every subscriber. Slow subscribers drop the

oldest event in their queue; we never block the publisher.

"""

async with self._lock:

subs = list(self._subscribers)

for q in subs:

try:

q.put_nowait(event)

except asyncio.QueueFull:

# Drop oldest, push newest. The slow subscriber is the

# one that loses an event, NOT the live primary path.

try:

_ = q.get_nowait()

except asyncio.QueueEmpty:

pass

try:

q.put_nowait(event)

except asyncio.QueueFull:

log.warning("subscriber queue still full after drop; event lost")

@asynccontextmanager

async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:

"""Async context manager — registers a queue, unregisters on exit."""

q: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=self._queue_max)

async with self._lock:

self._subscribers.add(q)

log.debug("subscriber registered, total=%d", len(self._subscribers))

try:

yield q

finally:

async with self._lock:

self._subscribers.discard(q)

log.debug("subscriber removed, remaining=%d", len(self._subscribers))

@property

def subscriber_count(self) -> int:

return len(self._subscribers)