BaseBroker

Message broker interface. All broker adapters (Inline, Redis Stream, Redis PubSub, etc.) implement this contract.

See Broker Adapters for concrete adapter configuration.

Base class for all broker implementations.

Provides shared behavior (UoW integration, connection recovery, capability checks, subscriber registration) via the Template Method pattern. Subclasses implement the abstract _underscore methods for broker-specific logic.

Source code in src/protean/port/broker.py

def __init__(
    self, name: str, domain: "Domain", conn_info: dict[str, str | bool]
) -> None:
    self.name = name
    self.domain = domain
    self.conn_info = conn_info

    self._subscribers: defaultdict[str, set[type[BaseSubscriber]]] = defaultdict(
        set
    )
    self._last_ping_time: float | None = None
    self._last_ping_success: bool | None = None
    self._start_time = time.time()

capabilities abstractmethod property

capabilities: BrokerCapabilities

Return the capabilities of this broker implementation.

RETURNS	DESCRIPTION
BrokerCapabilities	The capabilities supported by this broker TYPE: BrokerCapabilities

has_capability

has_capability(capability: BrokerCapabilities) -> bool

Check if broker has a specific capability.

PARAMETER	DESCRIPTION
capability	The capability to check for TYPE: BrokerCapabilities

RETURNS	DESCRIPTION
bool	True if the broker has the capability, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def has_capability(self, capability: BrokerCapabilities) -> bool:
    """Check if broker has a specific capability.

    Args:
        capability: The capability to check for

    Returns:
        bool: True if the broker has the capability, False otherwise
    """
    return capability in self.capabilities

has_all_capabilities

has_all_capabilities(
    capabilities: BrokerCapabilities,
) -> bool

Check if broker has all the specified capabilities.

PARAMETER	DESCRIPTION
capabilities	The capabilities to check for TYPE: BrokerCapabilities

RETURNS	DESCRIPTION
bool	True if the broker has all capabilities, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def has_all_capabilities(self, capabilities: BrokerCapabilities) -> bool:
    """Check if broker has all the specified capabilities.

    Args:
        capabilities: The capabilities to check for

    Returns:
        bool: True if the broker has all capabilities, False otherwise
    """
    return (self.capabilities & capabilities) == capabilities

has_any_capability

has_any_capability(
    capabilities: BrokerCapabilities,
) -> bool

Check if broker has any of the specified capabilities.

PARAMETER	DESCRIPTION
capabilities	The capabilities to check for TYPE: BrokerCapabilities

RETURNS	DESCRIPTION
bool	True if the broker has any of the capabilities, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def has_any_capability(self, capabilities: BrokerCapabilities) -> bool:
    """Check if broker has any of the specified capabilities.

    Args:
        capabilities: The capabilities to check for

    Returns:
        bool: True if the broker has any of the capabilities, False otherwise
    """
    return bool(self.capabilities & capabilities)

publish

publish(
    stream: str, message: dict[str, Any]
) -> Optional[str]

Publish a message to the broker.

PARAMETER	DESCRIPTION
stream	The stream to which the message should be published TYPE: str
message	The message payload to be published TYPE: dict

RETURNS	DESCRIPTION
str	The identifier of the message. The content of the identifier is broker-specific. TYPE: Optional[str]
Optional[str]	All brokers are guaranteed to provide message identifiers.

RAISES	DESCRIPTION
ValidationError	If message is an empty dict

Source code in src/protean/port/broker.py

def publish(self, stream: str, message: dict[str, Any]) -> Optional[str]:
    """Publish a message to the broker.

    Args:
        stream (str): The stream to which the message should be published
        message (dict): The message payload to be published

    Returns:
        str: The identifier of the message. The content of the identifier is broker-specific.
        All brokers are guaranteed to provide message identifiers.

    Raises:
        ValidationError: If message is an empty dict
    """
    if not message:
        raise ValidationError({"message": ["Message cannot be empty"]})

    if current_uow:
        logger.debug(f"Recording message {message} in {current_uow} for dispatch")

        current_uow.register_message(stream, message, broker_name=self.name)
        return None
    else:
        try:
            identifier = self._publish(stream, message)
        except Exception as e:
            # Check if this is a connection-related error and attempt recovery
            if self._is_connection_error(e):
                logger.warning(f"Connection error during publish: {e}")
                if self._ensure_connection():
                    # Retry the operation once after reconnection
                    identifier = self._publish(stream, message)
                else:
                    raise
            else:
                raise

        if (
            self.domain.config["message_processing"] == Processing.SYNC.value
            and self._subscribers[stream]
        ):
            for subscriber_cls in self._subscribers[stream]:
                subscriber = subscriber_cls()
                subscriber(message)

        return identifier

ping

ping() -> bool

Test broker connectivity.

RETURNS	DESCRIPTION
bool	True if broker is reachable and responsive, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def ping(self) -> bool:
    """Test broker connectivity.

    Returns:
        bool: True if broker is reachable and responsive, False otherwise
    """
    try:
        start_time = time.time()
        result = self._ping()
        self._last_ping_time = time.time() - start_time
        self._last_ping_success = result
        return result
    except Exception as e:
        logger.debug(f"Ping failed for broker {self.name}: {e}")
        self._last_ping_time = None
        self._last_ping_success = False
        return False

health_stats

health_stats() -> dict[str, Any]

Get comprehensive health statistics for the broker.

RETURNS	DESCRIPTION
dict	Health statistics with the following structure: { 'status': 'healthy' | 'degraded' | 'unhealthy', 'connected': bool, 'last_ping_ms': float | None, 'uptime_seconds': float, 'details': dict # Broker-specific details } TYPE: dict[str, Any]

Source code in src/protean/port/broker.py

def health_stats(self) -> dict[str, Any]:
    """Get comprehensive health statistics for the broker.

    Returns:
        dict: Health statistics with the following structure:
            {
                'status': 'healthy' | 'degraded' | 'unhealthy',
                'connected': bool,
                'last_ping_ms': float | None,
                'uptime_seconds': float,
                'details': dict  # Broker-specific details
            }
    """
    try:
        # Get broker-specific health details
        broker_details = self._health_stats()

        # Perform a fresh ping to get current connectivity status
        is_connected = self.ping()

        # Determine overall health status
        if is_connected and broker_details.get("healthy", True):
            status = "healthy"
        elif is_connected:
            status = "degraded"  # Connected but some issues reported
        else:
            status = "unhealthy"

        # Calculate uptime since broker initialization
        uptime_seconds = time.time() - self._start_time

        return {
            "status": status,
            "connected": is_connected,
            "last_ping_ms": self._last_ping_time * 1000
            if self._last_ping_time is not None
            else None,
            "uptime_seconds": uptime_seconds,
            "details": broker_details,
        }
    except Exception as e:
        logger.error(f"Error gathering health stats for broker {self.name}: {e}")
        return {
            "status": "unhealthy",
            "connected": False,
            "last_ping_ms": None,
            "uptime_seconds": 0,
            "details": {"error": str(e)},
        }

ensure_connection

ensure_connection() -> bool

Ensure broker connection is healthy, attempt reconnection if needed.

This method can be called explicitly or is triggered automatically when connection-related exceptions are encountered.

RETURNS	DESCRIPTION
bool	True if connection is healthy/restored, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def ensure_connection(self) -> bool:
    """Ensure broker connection is healthy, attempt reconnection if needed.

    This method can be called explicitly or is triggered automatically
    when connection-related exceptions are encountered.

    Returns:
        bool: True if connection is healthy/restored, False otherwise
    """
    return self._ensure_connection()

get_next

get_next(
    stream: str, consumer_group: str
) -> tuple[str, dict[str, Any]] | None

Retrieve the next message to process from broker.

PARAMETER	DESCRIPTION
stream	The stream from which to retrieve the message TYPE: str
consumer_group	The consumer group identifier TYPE: str

RETURNS	DESCRIPTION
tuple[str, dict[str, Any]] | None	tuple[str, dict] | None: A tuple of (identifier, message) or None if no messages available

Source code in src/protean/port/broker.py

def get_next(
    self, stream: str, consumer_group: str
) -> tuple[str, dict[str, Any]] | None:
    """Retrieve the next message to process from broker.

    Args:
        stream (str): The stream from which to retrieve the message
        consumer_group (str): The consumer group identifier

    Returns:
        tuple[str, dict] | None: A tuple of (identifier, message) or None if no messages available
    """
    # Check if broker supports consumer groups
    if not self.has_capability(BrokerCapabilities.CONSUMER_GROUPS):
        logger.warning(f"Broker {self.name} does not support consumer groups")
        return None

    try:
        return self._get_next(stream, consumer_group)
    except Exception as e:
        # Check if this is a connection-related error and attempt recovery
        if self._is_connection_error(e):
            logger.warning(f"Connection error during get_next: {e}")
            if self._ensure_connection():
                # Retry the operation once after reconnection
                return self._get_next(stream, consumer_group)
            else:
                raise
        else:
            raise

read

read(
    stream: str, consumer_group: str, no_of_messages: int
) -> list[tuple[str, dict[str, Any]]]

Read messages from the broker.

PARAMETER	DESCRIPTION
stream	The stream from which to read messages TYPE: str
consumer_group	The consumer group identifier TYPE: str
no_of_messages	The number of messages to read TYPE: int

RETURNS	DESCRIPTION
list[tuple[str, dict[str, Any]]]	list[tuple[str, dict]]: The list of (identifier, message) tuples

Source code in src/protean/port/broker.py

def read(
    self, stream: str, consumer_group: str, no_of_messages: int
) -> list[tuple[str, dict[str, Any]]]:
    """Read messages from the broker.

    Args:
        stream (str): The stream from which to read messages
        consumer_group (str): The consumer group identifier
        no_of_messages (int): The number of messages to read

    Returns:
        list[tuple[str, dict]]: The list of (identifier, message) tuples
    """
    # Check if broker supports consumer groups
    if not self.has_capability(BrokerCapabilities.CONSUMER_GROUPS):
        logger.warning(f"Broker {self.name} does not support consumer groups")
        return []

    try:
        return self._read(stream, consumer_group, no_of_messages)
    except Exception as e:
        # Check if this is a connection-related error and attempt recovery
        if self._is_connection_error(e):
            logger.warning(f"Connection error during read: {e}")
            if self._ensure_connection():
                # Retry the operation once after reconnection
                return self._read(stream, consumer_group, no_of_messages)
            else:
                raise
        else:
            raise

ack

ack(
    stream: str, identifier: str, consumer_group: str
) -> bool

Acknowledge successful processing of a message.

PARAMETER	DESCRIPTION
stream	The stream from which the message was received TYPE: str
identifier	The unique identifier of the message to acknowledge TYPE: str
consumer_group	The consumer group that processed the message TYPE: str

RETURNS	DESCRIPTION
bool	True if the message was successfully acknowledged, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def ack(self, stream: str, identifier: str, consumer_group: str) -> bool:
    """Acknowledge successful processing of a message.

    Args:
        stream (str): The stream from which the message was received
        identifier (str): The unique identifier of the message to acknowledge
        consumer_group (str): The consumer group that processed the message

    Returns:
        bool: True if the message was successfully acknowledged, False otherwise
    """
    # Check if broker supports acknowledgment
    if not self.has_capability(BrokerCapabilities.ACK_NACK):
        logger.warning(
            f"Broker {self.name} does not support message acknowledgment"
        )
        return False

    try:
        return self._ack(stream, identifier, consumer_group)
    except Exception as e:
        # Check if this is a connection-related error and attempt recovery
        if self._is_connection_error(e):
            logger.warning(f"Connection error during ack: {e}")
            if self._ensure_connection():
                # Retry the operation once after reconnection
                return self._ack(stream, identifier, consumer_group)
            else:
                raise
        else:
            raise

nack

nack(
    stream: str, identifier: str, consumer_group: str
) -> bool

Negative acknowledge - mark message for reprocessing.

PARAMETER	DESCRIPTION
stream	The stream from which the message was received TYPE: str
identifier	The unique identifier of the message to nack TYPE: str
consumer_group	The consumer group that failed to process the message TYPE: str

RETURNS	DESCRIPTION
bool	True if the message was successfully marked for reprocessing, False otherwise TYPE: bool

Source code in src/protean/port/broker.py

def nack(self, stream: str, identifier: str, consumer_group: str) -> bool:
    """Negative acknowledge - mark message for reprocessing.

    Args:
        stream (str): The stream from which the message was received
        identifier (str): The unique identifier of the message to nack
        consumer_group (str): The consumer group that failed to process the message

    Returns:
        bool: True if the message was successfully marked for reprocessing, False otherwise
    """
    # Check if broker supports negative acknowledgment
    if not self.has_capability(BrokerCapabilities.ACK_NACK):
        logger.warning(
            f"Broker {self.name} does not support message negative acknowledgment"
        )
        return False

    try:
        return self._nack(stream, identifier, consumer_group)
    except Exception as e:
        # Check if this is a connection-related error and attempt recovery
        if self._is_connection_error(e):
            logger.warning(f"Connection error during nack: {e}")
            if self._ensure_connection():
                # Retry the operation once after reconnection
                return self._nack(stream, identifier, consumer_group)
            else:
                raise
        else:
            raise

read_blocking

read_blocking(
    stream: str,
    consumer_group: str,
    consumer_name: str,
    timeout_ms: int = 5000,
    count: int = 1,
) -> list[tuple[str, dict[str, Any]]]

Read messages from the broker using blocking mode.

This is an optional method that brokers can implement to support efficient blocking reads for stream-based subscriptions.

PARAMETER	DESCRIPTION
stream	The stream from which to read messages TYPE: str
consumer_group	The consumer group identifier TYPE: str
consumer_name	The unique consumer name within the group TYPE: str
timeout_ms	Timeout in milliseconds to wait for messages (0 = block indefinitely) TYPE: int DEFAULT: 5000
count	Maximum number of messages to read TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
list[tuple[str, dict[str, Any]]]	list[tuple[str, dict]]: The list of (identifier, message) tuples

Source code in src/protean/port/broker.py

def read_blocking(
    self,
    stream: str,
    consumer_group: str,
    consumer_name: str,
    timeout_ms: int = 5000,
    count: int = 1,
) -> list[tuple[str, dict[str, Any]]]:
    """Read messages from the broker using blocking mode.

    This is an optional method that brokers can implement to support
    efficient blocking reads for stream-based subscriptions.

    Args:
        stream (str): The stream from which to read messages
        consumer_group (str): The consumer group identifier
        consumer_name (str): The unique consumer name within the group
        timeout_ms (int): Timeout in milliseconds to wait for messages (0 = block indefinitely)
        count (int): Maximum number of messages to read

    Returns:
        list[tuple[str, dict]]: The list of (identifier, message) tuples
    """
    # Check if broker supports blocking reads
    if not self.has_capability(BrokerCapabilities.BLOCKING_READ):
        # Fall back to regular read for brokers that don't support blocking
        return self._read(stream, consumer_group, count)

    try:
        return self._read_blocking(
            stream, consumer_group, consumer_name, timeout_ms, count
        )
    except Exception as e:
        # Check if this is a connection-related error and attempt recovery
        if self._is_connection_error(e):
            logger.warning(f"Connection error during read_blocking: {e}")
            if self._ensure_connection():
                # Retry the operation once after reconnection
                return self._read_blocking(
                    stream, consumer_group, consumer_name, timeout_ms, count
                )
            else:
                raise
        else:
            raise

info

info() -> dict[str, Any]

Get information about consumer groups and consumers in each group.

RETURNS	DESCRIPTION
dict	Information about consumer groups and their consumers TYPE: dict[str, Any]

Source code in src/protean/port/broker.py

def info(self) -> dict[str, Any]:
    """Get information about consumer groups and consumers in each group.

    Returns:
        dict: Information about consumer groups and their consumers
    """
    return self._info()

dlq_list

dlq_list(
    dlq_streams: list[str], limit: int = 100
) -> list[DLQEntry]

List DLQ messages across specified DLQ streams.

PARAMETER	DESCRIPTION
dlq_streams	List of DLQ stream names to query. TYPE: list[str]
limit	Maximum total messages to return. TYPE: int DEFAULT: 100

RETURNS	DESCRIPTION
list[DLQEntry]	List of DLQEntry objects sorted by failure time (newest first).

Source code in src/protean/port/broker.py

def dlq_list(self, dlq_streams: list[str], limit: int = 100) -> list[DLQEntry]:
    """List DLQ messages across specified DLQ streams.

    Args:
        dlq_streams: List of DLQ stream names to query.
        limit: Maximum total messages to return.

    Returns:
        List of DLQEntry objects sorted by failure time (newest first).
    """
    if not self.has_capability(BrokerCapabilities.DEAD_LETTER_QUEUE):
        logger.warning(f"Broker {self.name} does not support DLQ management")
        return []
    return self._dlq_list(dlq_streams, limit)

dlq_inspect

dlq_inspect(
    dlq_stream: str, dlq_id: str
) -> DLQEntry | None

Inspect a specific DLQ message by its DLQ entry ID.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream to look in. TYPE: str
dlq_id	The entry identifier within the DLQ stream. TYPE: str

RETURNS	DESCRIPTION
DLQEntry | None	DLQEntry if found, None otherwise.

Source code in src/protean/port/broker.py

def dlq_inspect(self, dlq_stream: str, dlq_id: str) -> DLQEntry | None:
    """Inspect a specific DLQ message by its DLQ entry ID.

    Args:
        dlq_stream: The DLQ stream to look in.
        dlq_id: The entry identifier within the DLQ stream.

    Returns:
        DLQEntry if found, None otherwise.
    """
    if not self.has_capability(BrokerCapabilities.DEAD_LETTER_QUEUE):
        logger.warning(f"Broker {self.name} does not support DLQ management")
        return None
    return self._dlq_inspect(dlq_stream, dlq_id)

dlq_replay

dlq_replay(
    dlq_stream: str, dlq_id: str, target_stream: str
) -> bool

Replay a single DLQ message back to its original stream.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream the message is in. TYPE: str
dlq_id	The entry identifier within the DLQ stream. TYPE: str
target_stream	The stream to re-publish the message to. TYPE: str

RETURNS	DESCRIPTION
bool	True if replayed successfully, False otherwise.

Source code in src/protean/port/broker.py

def dlq_replay(self, dlq_stream: str, dlq_id: str, target_stream: str) -> bool:
    """Replay a single DLQ message back to its original stream.

    Args:
        dlq_stream: The DLQ stream the message is in.
        dlq_id: The entry identifier within the DLQ stream.
        target_stream: The stream to re-publish the message to.

    Returns:
        True if replayed successfully, False otherwise.
    """
    if not self.has_capability(BrokerCapabilities.DEAD_LETTER_QUEUE):
        logger.warning(f"Broker {self.name} does not support DLQ management")
        return False
    return self._dlq_replay(dlq_stream, dlq_id, target_stream)

dlq_replay_all

dlq_replay_all(dlq_stream: str, target_stream: str) -> int

Replay all DLQ messages from a stream back to the original stream.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream to drain. TYPE: str
target_stream	The stream to re-publish messages to. TYPE: str

RETURNS	DESCRIPTION
int	Number of messages replayed.

Source code in src/protean/port/broker.py

def dlq_replay_all(self, dlq_stream: str, target_stream: str) -> int:
    """Replay all DLQ messages from a stream back to the original stream.

    Args:
        dlq_stream: The DLQ stream to drain.
        target_stream: The stream to re-publish messages to.

    Returns:
        Number of messages replayed.
    """
    if not self.has_capability(BrokerCapabilities.DEAD_LETTER_QUEUE):
        logger.warning(f"Broker {self.name} does not support DLQ management")
        return 0
    return self._dlq_replay_all(dlq_stream, target_stream)

dlq_purge

dlq_purge(dlq_stream: str) -> int

Purge all messages from a DLQ stream.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream to purge. TYPE: str

RETURNS	DESCRIPTION
int	Number of messages purged.

Source code in src/protean/port/broker.py

def dlq_purge(self, dlq_stream: str) -> int:
    """Purge all messages from a DLQ stream.

    Args:
        dlq_stream: The DLQ stream to purge.

    Returns:
        Number of messages purged.
    """
    if not self.has_capability(BrokerCapabilities.DEAD_LETTER_QUEUE):
        logger.warning(f"Broker {self.name} does not support DLQ management")
        return 0
    return self._dlq_purge(dlq_stream)

dlq_trim

dlq_trim(dlq_stream: str, min_id: str) -> int

Trim DLQ messages older than min_id (time-based trimming).

This is an optional method used by the DLQ maintenance task to remove messages that have exceeded their retention period. Brokers that don't support time-based trimming can leave the default implementation which returns 0.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream to trim. TYPE: str
min_id	Cutoff identifier (format varies by broker). Messages older than this cutoff will be removed. TYPE: str

RETURNS	DESCRIPTION
int	Number of messages trimmed.

Source code in src/protean/port/broker.py

def dlq_trim(self, dlq_stream: str, min_id: str) -> int:
    """Trim DLQ messages older than *min_id* (time-based trimming).

    This is an optional method used by the DLQ maintenance task to
    remove messages that have exceeded their retention period.
    Brokers that don't support time-based trimming can leave the
    default implementation which returns 0.

    Args:
        dlq_stream: The DLQ stream to trim.
        min_id: Cutoff identifier (format varies by broker).
            Messages older than this cutoff will be removed.

    Returns:
        Number of messages trimmed.
    """
    return 0

dlq_depth

dlq_depth(dlq_stream: str) -> int

Return the current number of messages in a DLQ stream.

Brokers that don't support depth queries return 0.

PARAMETER	DESCRIPTION
dlq_stream	The DLQ stream to query. TYPE: str

RETURNS	DESCRIPTION
int	Number of messages in the DLQ stream.

Source code in src/protean/port/broker.py

def dlq_depth(self, dlq_stream: str) -> int:
    """Return the current number of messages in a DLQ stream.

    Brokers that don't support depth queries return 0.

    Args:
        dlq_stream: The DLQ stream to query.

    Returns:
        Number of messages in the DLQ stream.
    """
    return 0

close

close() -> None

Close the broker and release all connections.

Subclasses that hold external resources (connection pools, sockets, etc.) should override this to perform cleanup. The default implementation is a no-op so that adapters without external resources (e.g. the inline broker) work without changes.

Source code in src/protean/port/broker.py

def close(self) -> None:
    """Close the broker and release all connections.

    Subclasses that hold external resources (connection pools, sockets,
    etc.) should override this to perform cleanup.  The default
    implementation is a no-op so that adapters without external
    resources (e.g. the inline broker) work without changes.
    """

register

register(subscriber_cls: Type[BaseSubscriber]) -> None

Register a subscriber to this broker against its stream.

PARAMETER	DESCRIPTION
subscriber_cls	The subscriber class connected to the stream. TYPE: Type[BaseSubscriber]

Source code in src/protean/port/broker.py

def register(self, subscriber_cls: Type[BaseSubscriber]) -> None:
    """Register a subscriber to this broker against its stream.

    Args:
        subscriber_cls: The subscriber class connected to the stream.
    """
    stream = subscriber_cls.meta_.stream

    self._subscribers[stream].add(subscriber_cls)

    logger.debug(
        f"Broker {self.name}: Registered Subscriber {subscriber_cls.__name__} for stream {stream}"
    )