Source code for simvx.core.event_bus

"""Typed event bus keyed on dataclass event types.

Handlers subscribe against a *dataclass* event type. Calling
:meth:`EventBus.publish` with an instance of that dataclass synchronously
fans the event out to every live handler subscribed for the *exact* class.
There is no MRO walk: a handler for ``Parent`` does NOT receive publishes
of ``Child(Parent)``; subclasses must subscribe themselves. This keeps
dispatch predictable and O(1) per event class.

Vocabulary (subscribe/publish/unsubscribe) is intentionally distinct from
:class:`Signal`'s connect/emit/disconnect: an ``EventBus`` is a typed pub/sub
hub, while a Signal is a per-instance callback list. Different shapes, so
different verbs.

Each event class has its own :class:`Signal`, so a handler's lifetime follows
the signal rule:

* A bound method of a :class:`Node` is held weakly, and its subscription ends
  when the node is destroyed (:meth:`Node.destroy`, or ``change_scene``, which
  destroys the scene it replaces) or garbage-collected. Leaving the tree
  without being destroyed (``remove_child``) does not end it.
* Every other callable (a lambda, a closure, a :class:`functools.partial`, a
  free function, a method of a plain object) is held strongly and fires until
  it is unsubscribed, even when it captures a node that was destroyed.

Two dispatch modes:

* :meth:`publish`: synchronous. Handlers run in subscription order on the
  calling thread before ``publish`` returns.
* :meth:`publish_deferred`: queues the event. The queue is drained when
  :meth:`flush_deferred` is called (typically once per frame).

Threading: the bus assumes a single-threaded engine. There are no locks.
Calling ``publish`` from a non-main thread is undefined behaviour.
"""

from __future__ import annotations

import dataclasses
from collections.abc import Callable
from typing import Any

from .signals import Signal

EventCls = type
Handler = Callable[[Any], None]




class EventBus:
    """Typed publish/subscribe bus keyed on dataclass event types."""

    __slots__ = ("_subs", "_deferred")

    def __init__(self) -> None:
        self._subs: dict[EventCls, Signal] = {}
        self._deferred: list[Any] = []

    # ------------------------------------------------------------------
    # Subscription
    # ------------------------------------------------------------------


    def subscribe(self, event_cls: EventCls, handler: Handler) -> None:
        """Register ``handler`` for events of exactly ``event_cls``.

        A bound method of a :class:`Node` is held weakly and dropped when the
        node is destroyed or collected; any other callable is held until
        :meth:`unsubscribe`.
        Subscribing a handler that is already subscribed does nothing.
        """
        if not callable(handler):
            raise TypeError(f"EventBus handler must be callable, got {type(handler).__name__!r}")
        signal = self._subs.get(event_cls)
        if signal is None:
            signal = self._subs[event_cls] = Signal(event_cls)
        elif signal._connection_for(handler) is not None:
            return
        signal.connect(handler)




    def unsubscribe(self, event_cls: EventCls, handler: Handler) -> None:
        """Remove ``handler`` from ``event_cls``. Idempotent.

        ``handler`` matches by equality, so a freshly bound method finds the
        subscription made with an earlier binding of the same method.
        """
        signal = self._subs.get(event_cls)
        if signal is None:
            return
        signal.disconnect(handler)
        if not signal._callbacks:
            del self._subs[event_cls]




    def disconnect_from_module(self, module_name: str) -> int:
        """Drop every handler defined in ``module_name``; return how many.

        Hot reload calls this before reloading a module, so the old version of
        a handler stops firing and the old module can be collected.
        """
        dropped = 0
        for event_cls, signal in list(self._subs.items()):
            dropped += signal.disconnect_from_module(module_name)
            if not signal._callbacks:
                del self._subs[event_cls]
        return dropped


    # ------------------------------------------------------------------
    # Dispatch
    # ------------------------------------------------------------------


    def publish(self, event: Any) -> None:
        """Synchronously fan ``event`` out to handlers of its exact type.

        Raises :class:`TypeError` if ``event`` is not a dataclass instance.
        Handlers run in subscription order. Subscriptions of collected nodes
        are pruned during dispatch.

        Unsubscribing takes effect immediately, as it does for a
        :class:`Signal`: a handler removed by an earlier handler in this same
        dispatch is not called. Unsubscribing mid-dispatch nearly always means
        the subscriber is being torn down, so calling it anyway is wrong.
        """
        if not dataclasses.is_dataclass(event) or isinstance(event, type):
            raise TypeError(f"EventBus.publish requires a dataclass instance, got {type(event).__name__!r}")
        event_cls = type(event)
        signal = self._subs.get(event_cls)
        if signal is None:
            return
        signal.emit(event)
        if not signal._callbacks and self._subs.get(event_cls) is signal:
            del self._subs[event_cls]




    def publish_deferred(self, event: Any) -> None:
        """Queue ``event`` for the next :meth:`flush_deferred` call."""
        if not dataclasses.is_dataclass(event) or isinstance(event, type):
            raise TypeError(f"EventBus.publish_deferred requires a dataclass instance, got {type(event).__name__!r}")
        self._deferred.append(event)




    def flush_deferred(self) -> None:
        """Drain the deferred queue, dispatching each event via :meth:`publish`."""
        if not self._deferred:
            return
        # Swap-and-clear so handlers that publish_deferred during flush land
        # on the next flush, not the current one.
        pending = self._deferred
        self._deferred = []
        for event in pending:
            self.publish(event)