jmri_core.events

 1"""Lightweight async-friendly signal/slot event bus.
 2
 3Replaces JMRI's ``PropertyChangeSupport`` / ``PropertyChangeListener``
 4machinery. A :class:`Signal` is a fan-out of zero or more listeners; emitting
 5delivers the payload to every listener synchronously, awaiting any that
 6return a coroutine.
 7
 8Design choices:
 9
10* No global event bus — every emitter owns its signals as instance attributes.
11  This keeps coupling explicit and lets tests construct beans in isolation.
12* Listeners are weakly *not* held — callers are responsible for the lifetime
13  of their callbacks (return value of :meth:`Signal.connect` is a disconnect
14  function). Keeping a strong reference avoids the surprise behaviour where
15  bound-method listeners disappear because nothing else holds the receiver.
16* Errors in one listener don't suppress delivery to the rest; the first
17  exception is collected and re-raised after every listener has run, mirroring
18  asyncio.gather semantics.
19"""
20from __future__ import annotations
21
22import asyncio
23import contextlib
24import inspect
25from collections.abc import Awaitable, Callable
26
27type Listener[T] = Callable[[T], None] | Callable[[T], Awaitable[None]]
28
29
30class Signal[T]:
31    """An emitter with zero or more listeners. See module docstring."""
32
33    __slots__ = ("_listeners",)
34
35    def __init__(self) -> None:
36        self._listeners: list[Listener[T]] = []
37
38    def connect(self, listener: Listener[T]) -> Callable[[], None]:
39        """Register *listener*. Returns a callable that disconnects it.
40
41        ``listener`` may be a sync or async callable taking one argument.
42        Returns a zero-arg disconnect function.
43        """
44        self._listeners.append(listener)
45
46        def disconnect() -> None:
47            # Idempotent: removing twice is a no-op.
48            with contextlib.suppress(ValueError):
49                self._listeners.remove(listener)
50
51        return disconnect
52
53    def disconnect_all(self) -> None:
54        """Drop every listener. Useful in teardown."""
55        self._listeners.clear()
56
57    @property
58    def listener_count(self) -> int:
59        return len(self._listeners)
60
61    async def emit(self, payload: T) -> None:
62        """Deliver *payload* to every connected listener.
63
64        Sync listeners run inline. Async listeners are awaited sequentially in
65        registration order — emission preserves ordering, which beans rely on
66        for state-change semantics (commanded-state notifications must arrive
67        before known-state notifications). If you need concurrency, use
68        :meth:`emit_concurrent`.
69        """
70        first_error: BaseException | None = None
71        # Snapshot the list so listeners that disconnect during dispatch don't
72        # mutate the iteration.
73        for listener in list(self._listeners):
74            try:
75                result = listener(payload)
76                if inspect.iscoroutine(result):
77                    await result
78            except BaseException as e:  # noqa: BLE001 — re-raised below
79                if first_error is None:
80                    first_error = e
81        if first_error is not None:
82            raise first_error
83
84    async def emit_concurrent(self, payload: T) -> None:
85        """Fan out to every async listener concurrently via ``asyncio.gather``.
86
87        Sync listeners still run inline before dispatching the async batch.
88        Use only when listeners are explicitly independent.
89        """
90        async_results: list[Awaitable[None]] = []
91        for listener in list(self._listeners):
92            result = listener(payload)
93            if inspect.iscoroutine(result):
94                async_results.append(result)
95        if async_results:
96            await asyncio.gather(*async_results)