Skip to content

Message Bus Command Sender

message_bus_command_sender

Message-bus-backed CommandSenderPort adapter.

Delegates command dispatch to an injected MessageBusPort.

MessageBusCommandSender

Bases: CommandSenderPort[CommandPayloadType]

Infrastructure adapter that sends commands via a MessageBusPort.

Implements CommandSenderPort by delegating send to
MessageBusPort.dispatch.

Example 
class Command[T]:
    def __init__(self) -> None:
        pass


class InMemoryMessageBus[MT, R]:
    async def dispatch(self, message: MT) -> R: ...
    def register(self, message_type, handler): ...


class MyCommand(Command[dict[str, object]]):
    def __init__(self) -> None:
        pass


bus = InMemoryMessageBus[MyCommand, None]()
sender = MessageBusCommandSender[dict[str, object]](bus)
await sender.send(MyCommand())
Source code in src/forging_blocks/infrastructure/message_bus/message_bus_command_sender.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class MessageBusCommandSender[CommandPayloadType](CommandSenderPort[CommandPayloadType]):
    """Infrastructure adapter that sends commands via a ``MessageBusPort``.

    Implements ``CommandSenderPort`` by delegating ``send`` to
    ``MessageBusPort.dispatch``.

    Example:
        ```python
        class Command[T]:
            def __init__(self) -> None:
                pass


        class InMemoryMessageBus[MT, R]:
            async def dispatch(self, message: MT) -> R: ...
            def register(self, message_type, handler): ...


        class MyCommand(Command[dict[str, object]]):
            def __init__(self) -> None:
                pass


        bus = InMemoryMessageBus[MyCommand, None]()
        sender = MessageBusCommandSender[dict[str, object]](bus)
        await sender.send(MyCommand())
        ```
    """

    def __init__(self, message_bus: MessageBusPort[Command[CommandPayloadType], None]) -> None:
        self._message_bus = message_bus

    async def send(self, command: Command[CommandPayloadType]) -> None:
        """Send a command asynchronously via the message bus."""
        await self._message_bus.dispatch(command)

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

Send a command asynchronously via the message bus.

Source code in src/forging_blocks/infrastructure/message_bus/message_bus_command_sender.py 
43
44
45
async def send(self, command: Command[CommandPayloadType]) -> None:
    """Send a command asynchronously via the message bus."""
    await self._message_bus.dispatch(command)