budosystems.events.message_bus.MessageBus

class MessageBus

Bases: object

The event message bus.

This is a singleton class. The canonical reference to the instance is budosystems.events.MESSAGE_BUS.

static __new__(cls) → MessageBus
register_event_handler(event_type: Type[Event], handler: AsyncEventHandler[Event]) → None

Registers a callable as a handler for a specific event type.

Parameters:

  • event_type – A subclass of Event.

  • handler – An async callable which will handle instances of event_type.

get_event_handlers(*, event_type: Type[EventType_contra] | None = None, handler: AsyncEventHandler[EventType_contra] | None = None, all_handlers: bool = False) → Set[Registration[Event]]

Gets a filtered list of registered event handlers.

If only one of event_type or handler is specified, includes all handlers associated with that parameter.

If both event_type and handler are specified:

  • if all_handlers is False (default), the returned list includes the associated pair only if it was previously registered.

  • if all_handlers is True, the returned list includes all registrations that match either event_type or handler.

When event_type is specified, the returned list only includes registrations that match the type exactly (i.e. no super- or sub-types).

If neither event_type nor handler is specified, and all_handlers is True, the returned list will include all registrations.

Parameters:

  • event_type – A subclass of Event.

  • handler – An async callable which handled instances of event_type.

  • all_handlers – A flag to determine whether or not to include everything.

Returns:

A set of (event_type, handler) pairs of registered event handlers.

unregister_event_handlers(*, event_type: Type[EventType_contra] | None = None, handler: AsyncEventHandler[EventType_contra] | None = None, all_handlers: bool = False) → None

Removes event handlers.

See get_event_handlers for the matching process.

Parameters:

  • event_type – A subclass of Event.

  • handler – An async callable which handled instances of event_type.

  • all_handlers – A flag to determine whether or not to include every thing. Only considered if the other two arguments are None (default).

signal(event: EventType_contra) → tuple[bool, list[Any]]

Calls the handlers for the incoming event. This includes handlers registered for supertypes of event.

Processing order of event handlers is not guaranteed. The only guarantee is that all eligible handlers will be processed when signal returns.

Parameters:

event – An event being signaled.