orchestrata_ultimate.postman.event_emitting_postman

Reference › orchestrata_ultimate › postman › event_emitting_postman

A decorator around Postman that publishes system events about routing, so observers can watch the broker without being coupled to it. The postman package initialiser re-exports it. It serves SPEC-002 story US-001 (REQ-002-US-001-001, -004).

EventEmittingPostman takes a wrapped Postman and an optional event bus implementing IEventBus; __getattr__ forwards any other attribute to the wrapped object. route calls the wrapped route and, when an event bus is set and the result succeeded, emits a postman.message.routed event with the packet id and destination. acknowledge emits postman.message.delivered, and retry_failed emits postman.message.failed with the error when the retry still fails. A failure to emit is logged and the delivery result is returned unchanged. The events go to the kernel's EventBus when that is the bus supplied.

At a glance

Class diagram

Dependency diagram

API

Classes

EventEmittingPostman

Decorators: traces(spec='002', story='US-001', reqs=['REQ-002-US-001-001']) · Line 15

Postman wrapper that emits system events for routing and delivery.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/event_emitting_postman.py

"""Event-emitting wrapper for Postman message routing."""

import logging
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import IEventBus, SystemEvent
from orchestrata_ultimate.postman.postman import DeliveryResult, Postman

logger = logging.getLogger(__name__)


@traces(spec="002", story="US-001", reqs=["REQ-002-US-001-001"])
class EventEmittingPostman:
    """Postman wrapper that emits system events for routing and delivery.

    Wraps a Postman instance and emits events to an optional IEventBus on
    route, acknowledge, and retry_failed operations. When event_bus is None
    the wrapper is a transparent pass-through.
    """

    def __init__(self, postman: Postman, event_bus: IEventBus | None = None) -> None:
        """Initialise the wrapper.

        Args:
            postman: The Postman instance to wrap.
            event_bus: Optional event bus for emission. No events are emitted
                if None.
        """
        self._postman = postman
        self._event_bus = event_bus

    def __getattr__(self, name: str) -> Any:
        """Delegate all other attribute access to the wrapped Postman."""
        return getattr(self._postman, name)

    async def route(self, packet: Any) -> DeliveryResult:
        """Route a packet and emit postman.message.routed on success.

        Args:
            packet: CXP packet to route.

        Returns:
            DeliveryResult from the wrapped Postman.
        """
        result: DeliveryResult = await self._postman.route(packet)
        if self._event_bus is not None and result.success:
            event = SystemEvent(
                type="postman.message.routed",
                source="postman",
                timestamp=datetime.now(UTC),
                payload={
                    "packet_id": result.packet_id,
                    "destination": result.destination,
                },
            )
            try:
                await self._event_bus.emit(event)
            except Exception:
                logger.exception("Failed to emit postman.message.routed")
        return result

    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-015"])
    async def acknowledge(self, packet_id: str) -> None:
        """Acknowledge delivery and emit postman.message.delivered.

        Args:
            packet_id: Packet ID to acknowledge.
        """
        await self._postman.acknowledge(packet_id)
        if self._event_bus is not None:
            event = SystemEvent(
                type="postman.message.delivered",
                source="postman",
                timestamp=datetime.now(UTC),
                payload={"packet_id": packet_id},
            )
            try:
                await self._event_bus.emit(event)
            except Exception:
                logger.exception("Failed to emit postman.message.delivered")

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-004"])
    async def retry_failed(self, packet_id: str) -> DeliveryResult:
        """Retry a failed packet and emit postman.message.failed if still failing.

        Args:
            packet_id: Packet ID to retry.

        Returns:
            DeliveryResult from the wrapped Postman.
        """
        result: DeliveryResult = await self._postman.retry_failed(packet_id)
        if self._event_bus is not None and not result.success:
            event = SystemEvent(
                type="postman.message.failed",
                source="postman",
                timestamp=datetime.now(UTC),
                payload={
                    "packet_id": packet_id,
                    "error": result.error or "unknown",
                },
            )
            try:
                await self._event_bus.emit(event)
            except Exception:
                logger.exception("Failed to emit postman.message.failed")
        return result