Skip to content

Event Bus Port

event_bus_port

Event bus port for publishing events and sending commands.

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

EventBusPort

Bases: ABC

Abstract event bus for publishing events and sending commands.

Implementations handle
  • Publishing events to one or more registered handlers.
  • Sending commands to a single registered handler.
  • Registering handlers for specific message types.
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
class EventBusPort[EventPayloadType, CommandPayloadType](ABC):
    """Abstract event bus for publishing events and sending commands.

    Implementations handle:
      - Publishing events to one or more registered handlers.
      - Sending commands to a single registered handler.
      - Registering handlers for specific message types.
    """

    @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, handler: Any) -> None:
        """Register a handler for the given message type.

        Args:
            message_type: The message class to handle.
            handler: A handler instance (``EventHandler`` or ``CommandHandler``).
        """
        ...

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 
26
27
28
29
30
31
32
33
34
35
36
@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 
38
39
40
41
42
43
44
45
46
47
48
@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, handler: Any) -> None abstractmethod

Register a handler for the given message type.

Parameters:

Name Type Description Default
message_type type

The message class to handle.

 required
handler Any

A handler instance (EventHandler or CommandHandler).

 required
Source code in src/forging_blocks/application/ports/outbound/event_bus_port.py 
50
51
52
53
54
55
56
57
58
@abstractmethod
def register_handler(self, message_type: type, handler: Any) -> None:
    """Register a handler for the given message type.

    Args:
        message_type: The message class to handle.
        handler: A handler instance (``EventHandler`` or ``CommandHandler``).
    """
    ...