import sys
|
from typing import Any, Awaitable, Callable, TypeVar
|
|
from frozenlist import FrozenList
|
|
if sys.version_info >= (3, 11):
|
from typing import Unpack
|
else:
|
from typing_extensions import Unpack
|
|
if sys.version_info >= (3, 13):
|
from typing import TypeVarTuple
|
else:
|
from typing_extensions import TypeVarTuple
|
|
_T = TypeVar("_T")
|
_Ts = TypeVarTuple("_Ts", default=Unpack[tuple[()]])
|
|
__version__ = "1.4.0"
|
|
__all__ = ("Signal",)
|
|
|
class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
|
"""Coroutine-based signal implementation.
|
|
To connect a callback to a signal, use any list method.
|
|
Signals are fired using the send() coroutine, which takes named
|
arguments.
|
"""
|
|
__slots__ = ("_owner",)
|
|
def __init__(self, owner: object):
|
super().__init__()
|
self._owner = owner
|
|
def __repr__(self) -> str:
|
return "<Signal owner={}, frozen={}, {!r}>".format(
|
self._owner, self.frozen, list(self)
|
)
|
|
async def send(self, *args: Unpack[_Ts], **kwargs: Any) -> None:
|
"""
|
Sends data to all registered receivers.
|
"""
|
if not self.frozen:
|
raise RuntimeError("Cannot send non-frozen signal.")
|
|
for receiver in self:
|
await receiver(*args, **kwargs)
|
|
def __call__(
|
self, func: Callable[[Unpack[_Ts]], Awaitable[_T]]
|
) -> Callable[[Unpack[_Ts]], Awaitable[_T]]:
|
"""Decorator to add a function to this Signal."""
|
self.append(func)
|
return func
|
