In Memory Event Bus

in_memory_event_bus

In-memory implementation of the EventBusPort port.

Dispatches events to multiple registered handlers (fan-out) and
commands to a single registered handler. Handlers are looked up
by the exact type of the message.

InMemoryEventBus

Bases: EventBusPort[EventPayloadType, CommandPayloadType]

In-memory event bus with separate event/command dispatch.

Attributes:

Name	Type	Description
_event_handlers	dict[type[Any], list[EventHandler[Any]]]	Per-event-type list of handlers.
_command_handlers	dict[type[Any], CommandHandler[Any]]	Per-command-type single handler.

Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py

class InMemoryEventBus[EventPayloadType, CommandPayloadType](
    EventBusPort[EventPayloadType, CommandPayloadType]
):
    """In-memory event bus with separate event/command dispatch.

    Attributes:
        _event_handlers: Per-event-type list of handlers.
        _command_handlers: Per-command-type single handler.
    """

    __slots__ = ("_command_handlers", "_event_handlers")

    def __init__(self) -> None:
        self._event_handlers: dict[type[Any], list[EventHandler[Any]]] = {}
        self._command_handlers: dict[type[Any], CommandHandler[Any]] = {}

    def register_handler(self, message_type: type[Any], handler: Any) -> None:
        """Register a handler for a message type.

        For event types, multiple handlers can be registered (fan-out).
        For command types, only one handler is allowed per type.

        Args:
            message_type: The message class to handle.
            handler: A handler instance.
        """
        if issubclass(message_type, Event):
            self._event_handlers.setdefault(cast(type[Any], message_type), []).append(handler)
        elif issubclass(message_type, Command):
            self._command_handlers[message_type] = handler

    async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
        """Publish an event to all registered handlers.

        Args:
            event: The domain event.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
            handler raises.
        """
        handlers = self._event_handlers.get(type(event), [])
        for handler in handlers:
            try:
                await handler.handle(event)
            except Exception as exc:
                return Err(EventBusError(str(exc)))
        return Ok(None)

    async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
        """Send a command to its registered handler.

        Args:
            command: The command.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
            handler raises or no handler is registered.
        """
        handler = self._command_handlers.get(type(command))
        if handler is None:
            return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
        try:
            await handler.handle(command)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
        return Ok(None)

register_handler(message_type: type[Any], handler: Any) -> None

Register a handler for a message type.

For event types, multiple handlers can be registered (fan-out).
For command types, only one handler is allowed per type.

Parameters:

Name	Type	Description	Default
message_type	type[Any]	The message class to handle.	required
handler	Any	A handler instance.	required

Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py

def register_handler(self, message_type: type[Any], handler: Any) -> None:
    """Register a handler for a message type.

    For event types, multiple handlers can be registered (fan-out).
    For command types, only one handler is allowed per type.

    Args:
        message_type: The message class to handle.
        handler: A handler instance.
    """
    if issubclass(message_type, Event):
        self._event_handlers.setdefault(cast(type[Any], message_type), []).append(handler)
    elif issubclass(message_type, Command):
        self._command_handlers[message_type] = handler

publish(event: Event[EventPayloadType]) -> Result[None, EventBusError] async

Publish an event to all registered handlers.

Parameters:

Name	Type	Description	Default
event	Event[EventPayloadType]	The domain event.	required

Returns:

Type	Description
Result[None, EventBusError]	Ok(None) on success, or Err(EventBusError) if any
Result[None, EventBusError]	handler raises.

Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py

async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
    """Publish an event to all registered handlers.

    Args:
        event: The domain event.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
        handler raises.
    """
    handlers = self._event_handlers.get(type(event), [])
    for handler in handlers:
        try:
            await handler.handle(event)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
    return Ok(None)

send(command: Command[CommandPayloadType]) -> Result[None, EventBusError] async

Send a command to its registered handler.

Parameters:

Name	Type	Description	Default
command	Command[CommandPayloadType]	The command.	required

Returns:

Type	Description
Result[None, EventBusError]	Ok(None) on success, or Err(EventBusError) if the
Result[None, EventBusError]	handler raises or no handler is registered.

Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py

async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
    """Send a command to its registered handler.

    Args:
        command: The command.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
        handler raises or no handler is registered.
    """
    handler = self._command_handlers.get(type(command))
    if handler is None:
        return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
    try:
        await handler.handle(command)
    except Exception as exc:
        return Err(EventBusError(str(exc)))
    return Ok(None)