Skip to content

Event Bus Port

event_bus_port

Event bus port for publishing events and sending commands.

Defines the EventBusPort contract for in-process message dispatch
with separate policies for events (multi-handler fan-out) and commands
(single-handler routing).

EventBusPort

Bases: OutboundPort

Abstract base class for event buses that publish events and send commands.

Responsibilities
  • Publish domain events to all registered handlers (fan-out).
  • Route commands to a single registered handler.
  • Manage handler registration and lookup by message type.
Non-Responsibilities
  • Guarantee delivery or implement transactional outbox.
  • Persist messages or provide replay capabilities.
  • Manage subscriptions, topics, or routing rules — those belong
    to infrastructure.
Example 
bus = MyEventBus[OrderData, PlaceOrderData, object]()
bus.register_handler(OrderShipped, ship_order_handler)
await bus.publish(OrderShipped(order_id="42"))
Source code in src/forging_blocks/application/ports/outbound/event_bus_port.py 
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class EventBusPort[EventPayloadType, CommandPayloadType, HandlerType](
    OutboundPort,
):
    """Abstract base class for event buses that publish events and send commands.

    Responsibilities:
        - Publish domain events to all registered handlers (fan-out).
        - Route commands to a single registered handler.
        - Manage handler registration and lookup by message type.

    Non-Responsibilities:
        - Guarantee delivery or implement transactional outbox.
        - Persist messages or provide replay capabilities.
        - Manage subscriptions, topics, or routing rules — those belong
          to infrastructure.

    Example:
        ```python
        bus = MyEventBus[OrderData, PlaceOrderData, object]()
        bus.register_handler(OrderShipped, ship_order_handler)
        await bus.publish(OrderShipped(order_id="42"))
        ```
    """

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

        Args:
            event: The domain event to publish.

        Returns:
            A ``Result`` indicating success or an ``EventBusError``.

        """
        ...

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

        Args:
            command: The command to dispatch.

        Returns:
            A ``Result`` indicating success or an ``EventBusError``.

        """
        ...

    @abstractmethod
    def register_handler(
        self,
        message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
        handler: HandlerType,
    ) -> None:
        """Register a handler for the given message type.

        Args:
            message_type: The message class to handle.
            handler: A handler instance implementing the handler contract.

        """

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

Publish a domain event to all registered handlers.

Parameters:

Name Type Description Default
event Event[EventPayloadType]

The domain event to publish.

 required

Returns:

Type Description
Result[None, EventBusError]

A Result indicating success or an EventBusError.
Source code in src/forging_blocks/application/ports/outbound/event_bus_port.py 
41
42
43
44
45
46
47
48
49
50
51
52
@abstractmethod
async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
    """Publish a domain event to all registered handlers.

    Args:
        event: The domain event to publish.

    Returns:
        A ``Result`` indicating success or an ``EventBusError``.

    """
    ...

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

Send a command to its registered handler.

Parameters:

Name Type Description Default
command Command[CommandPayloadType]

The command to dispatch.

 required

Returns:

Type Description
Result[None, EventBusError]

A Result indicating success or an EventBusError.
Source code in src/forging_blocks/application/ports/outbound/event_bus_port.py 
54
55
56
57
58
59
60
61
62
63
64
65
@abstractmethod
async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
    """Send a command to its registered handler.

    Args:
        command: The command to dispatch.

    Returns:
        A ``Result`` indicating success or an ``EventBusError``.

    """
    ...

register_handler(message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]], handler: HandlerType) -> None abstractmethod

Register a handler for the given message type.

Parameters:

Name Type Description Default
message_type type[Event[EventPayloadType]] | type[Command[CommandPayloadType]]

The message class to handle.

 required
handler HandlerType

A handler instance implementing the handler contract.

 required
Source code in src/forging_blocks/application/ports/outbound/event_bus_port.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
@abstractmethod
def register_handler(
    self,
    message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
    handler: HandlerType,
) -> None:
    """Register a handler for the given message type.

    Args:
        message_type: The message class to handle.
        handler: A handler instance implementing the handler contract.

    """