In Memory Event Bus

in_memory_event_bus

In-memory implementation of the EventBusPort port.

This implementation stores handlers in memory and is suitable for testing
and development purposes.

InMemoryEventBus

Bases: EventBusPort

In-memory implementation of the EventBusPort port.

Stores event handlers and command handlers in dictionaries.

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

class InMemoryEventBus(EventBusPort):
    """
    In-memory implementation of the EventBusPort port.

    Stores event handlers and command handlers in dictionaries.
    """

    def __init__(self) -> None:
        self._event_handlers: dict[Type[Event[Any]], List[EventHandler]] = defaultdict(list)
        self._command_handlers: dict[Type[Command[Any]], CommandHandler] = {}

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

        Args:
            event: The event to publish.
        """
        handlers = self._event_handlers.get(type(event), [])
        for handler in handlers:
            await handler(event)

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

        Args:
            command: The command to send.

        Raises:
            NoHandlerError: If no handler is registered for the command type.
        """
        handler = self._command_handlers.get(type(command))
        if handler is None:
            raise NoHandlerError(
                f"No handler registered for command type: {type(command).__name__}"
            )
        await handler(command)

    def subscribe(self, event_type: Type[Event[Any]], handler: EventHandler) -> None:
        """
        Subscribe a handler to an event type.

        Args:
            event_type: The type of event to subscribe to.
            handler: The handler function to call when the event is published.
        """
        self._event_handlers[event_type].append(handler)

    def register_command_handler(
        self, command_type: Type[Command[Any]], handler: CommandHandler
    ) -> None:
        """
        Register a handler for a command type.

        Args:
            command_type: The type of command to handle.
            handler: The handler function to call when the command is sent.
        """
        self._command_handlers[command_type] = handler

publish(event: Event[Any]) -> None async

Publish an event to all registered handlers.

Parameters:

Name	Type	Description	Default
event	Event[Any]	The event to publish.	required

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

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

    Args:
        event: The event to publish.
    """
    handlers = self._event_handlers.get(type(event), [])
    for handler in handlers:
        await handler(event)

send(command: Command[Any]) -> None async

Send a command to its handler.

Parameters:

Name	Type	Description	Default
command	Command[Any]	The command to send.	required

Raises:

Type	Description
NoHandlerError	If no handler is registered for the command type.

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

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

    Args:
        command: The command to send.

    Raises:
        NoHandlerError: If no handler is registered for the command type.
    """
    handler = self._command_handlers.get(type(command))
    if handler is None:
        raise NoHandlerError(
            f"No handler registered for command type: {type(command).__name__}"
        )
    await handler(command)

subscribe(event_type: Type[Event[Any]], handler: EventHandler) -> None

Subscribe a handler to an event type.

Parameters:

Name	Type	Description	Default
event_type	Type[Event[Any]]	The type of event to subscribe to.	required
handler	EventHandler	The handler function to call when the event is published.	required

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

def subscribe(self, event_type: Type[Event[Any]], handler: EventHandler) -> None:
    """
    Subscribe a handler to an event type.

    Args:
        event_type: The type of event to subscribe to.
        handler: The handler function to call when the event is published.
    """
    self._event_handlers[event_type].append(handler)

register_command_handler(command_type: Type[Command[Any]], handler: CommandHandler) -> None

Register a handler for a command type.

Parameters:

Name	Type	Description	Default
command_type	Type[Command[Any]]	The type of command to handle.	required
handler	CommandHandler	The handler function to call when the command is sent.	required

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

def register_command_handler(
    self, command_type: Type[Command[Any]], handler: CommandHandler
) -> None:
    """
    Register a handler for a command type.

    Args:
        command_type: The type of command to handle.
        handler: The handler function to call when the command is sent.
    """
    self._command_handlers[command_type] = handler