PipelineBridge

def __init__(self, maxsize: int = 1,
             on_drop: Optional[Callable[[], None]] = None) -> None:
    """Initializes the bridge.

    Args:
        maxsize: Queue depth; 1 means latest-wins, larger is drop-oldest.
        on_drop: Fired (on the loop thread) whenever a queued item is
            dropped. The video bridge uses it to force a recovery keyframe:
            a dropped ENCODED frame breaks the wire reference chain with no
            RTP gap, so the browser never requests a PLI and the smear
            would persist under infinite GOP.
    """
    self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
    self._on_drop = on_drop

def set_data(self, data: Any) -> None:
    """Enqueue an item, dropping the oldest one when the queue is full.

    Synchronous, no lock: the drop-oldest check and the put have no await,
    so the single-threaded loop runs them without interleaving (all access
    is on the loop thread). A full queue means the consumer is lagging, so
    the oldest queued item is dropped to make space for the new one.
    """
    if self._queue.full():
        self._queue.get_nowait()
        if self._on_drop is not None:
            self._on_drop()
    self._queue.put_nowait(data)

async def get_data(self) -> Any:
    """Wait until an item is available in the queue and return it."""
    return await self._queue.get()