musicmouse/python-backend/musicmouse/bus.py

"""The event bus.

Everything in the process is serialised through one FIFO queue on one loop, which is
what makes "last event wins" a well-defined rule for LED zone arbitration and what
makes scenario tests deterministic.

Handlers may be sync or async; async handlers are awaited, so one event is fully
handled before the next is dispatched. A handler that raises is logged and does not
stop the others or the bus.
"""

from __future__ import annotations

import asyncio
import contextlib
import inspect
import logging
from collections.abc import Callable, Coroutine
from typing import Any, TypeAlias, TypeVar

from musicmouse.events import Event

_log = logging.getLogger(__name__)

__all__ = ["EventBus", "Handler", "Unsubscribe"]

#: An alias carrying a TypeVar is generic on its own, so ``Handler[SomeEvent]`` still
#: parameterises it the way the PEP 695 form did.
E = TypeVar("E", bound=Event)

Handler: TypeAlias = Callable[[E], Coroutine[Any, Any, None] | None]
Unsubscribe: TypeAlias = Callable[[], None]


class EventBus:
    def __init__(self) -> None:
        self._handlers: dict[type[Event], list[Handler[Any]]] = {}
        self._wildcard: list[Handler[Any]] = []
        self._resolved: dict[type[Event], tuple[Handler[Any], ...]] = {}
        self._queue: asyncio.Queue[Event] = asyncio.Queue()
        self._loop: asyncio.AbstractEventLoop | None = None
        self._dispatcher: asyncio.Task[None] | None = None

    # ------------------------------------------------------------------ lifecycle

    async def start(self) -> None:
        if self._dispatcher is not None:
            return
        self._loop = asyncio.get_running_loop()
        self._dispatcher = asyncio.create_task(self._run(), name="event-bus")

    async def stop(self) -> None:
        if self._dispatcher is None:
            return
        self._dispatcher.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._dispatcher
        self._dispatcher = None
        self._loop = None

    async def __aenter__(self) -> EventBus:
        await self.start()
        return self

    async def __aexit__(self, *exc_info: object) -> None:
        await self.stop()

    # --------------------------------------------------------------- subscription

    def subscribe(self, event_type: type[E], handler: Handler[E]) -> Unsubscribe:
        """Register ``handler`` for ``event_type`` and any subclass of it."""
        self._handlers.setdefault(event_type, []).append(handler)
        self._resolved.clear()

        def unsubscribe() -> None:
            handlers = self._handlers.get(event_type)
            if handlers and handler in handlers:
                handlers.remove(handler)
                self._resolved.clear()

        return unsubscribe

    def subscribe_all(self, handler: Handler[Event]) -> Unsubscribe:
        """Register ``handler`` for every event. Useful for logging and broadcasting."""
        self._wildcard.append(handler)

        def unsubscribe() -> None:
            if handler in self._wildcard:
                self._wildcard.remove(handler)

        return unsubscribe

    # ---------------------------------------------------------------- publication

    def emit(self, event: Event) -> None:
        """Queue ``event`` for dispatch. Safe to call from any thread.

        libVLC fires its callbacks on its own thread; this is where that crossing is
        made safe instead of reaching the serial transport off-loop.
        """
        loop = self._loop
        if loop is None:
            raise RuntimeError("EventBus.emit() before start()")
        try:
            running = asyncio.get_running_loop()
        except RuntimeError:
            running = None
        if running is loop:
            self._queue.put_nowait(event)
        else:
            loop.call_soon_threadsafe(self._queue.put_nowait, event)

    async def drain(self) -> None:
        """Wait until every queued event, and everything they emitted, is handled."""
        await self._queue.join()

    async def emit_and_wait(self, event: Event) -> None:
        self.emit(event)
        await self.drain()

    # -------------------------------------------------------------------- internals

    async def _run(self) -> None:
        while True:
            event = await self._queue.get()
            try:
                await self._dispatch(event)
            finally:
                self._queue.task_done()

    async def _dispatch(self, event: Event) -> None:
        for handler in self._handlers_for(type(event)):
            try:
                result = handler(event)
                if inspect.isawaitable(result):
                    await result
            except asyncio.CancelledError:
                raise
            except Exception:
                _log.exception("Handler %s failed on %r", _name(handler), event)

    def _handlers_for(self, event_type: type[Event]) -> tuple[Handler[Any], ...]:
        cached = self._resolved.get(event_type)
        if cached is None:
            matched: list[Handler[Any]] = []
            for klass in event_type.__mro__:
                if klass is object:
                    continue
                matched.extend(self._handlers.get(klass, ()))
            cached = tuple(matched)
            self._resolved[event_type] = cached
        # Wildcards are not cached: they are appended last and change rarely.
        return cached + tuple(self._wildcard)


def _name(handler: Handler[Any]) -> str:
    return getattr(handler, "__qualname__", repr(handler))