Skip to content

Message Bus Event Publisher

message_bus_event_publisher

Message-bus-backed EventPublisherPort adapter.

Delegates event publishing to an injected MessageBusPort.

MessageBusEventPublisher

Bases: EventPublisherPort[EventPayloadType]

Infrastructure adapter that publishes events via a MessageBusPort.

Implements EventPublisherPort by delegating publish to
MessageBusPort.dispatch.

Source code in src/forging_blocks/infrastructure/message_bus/message_bus_event_publisher.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
class MessageBusEventPublisher[EventPayloadType](EventPublisherPort[EventPayloadType]):
    """Infrastructure adapter that publishes events via a ``MessageBusPort``.

    Implements ``EventPublisherPort`` by delegating ``publish`` to
    ``MessageBusPort.dispatch``.
    """

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

    async def publish(self, event: Event[EventPayloadType]) -> None:
        """Publish a domain event via the message bus."""
        await self._message_bus.dispatch(event)

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

Publish a domain event via the message bus.

Source code in src/forging_blocks/infrastructure/message_bus/message_bus_event_publisher.py 
21
22
23
async def publish(self, event: Event[EventPayloadType]) -> None:
    """Publish a domain event via the message bus."""
    await self._message_bus.dispatch(event)