Skip to main content
Version: Next

EventConsumer & AsyncLoopConsumer

Abstract Backend separates the consumer contract from the default runtime implementation so you can plug in alternative event loops when needed.

  • EventConsumer lives in abe/backends/message_queue/base/consumer.py and is defined as a PEP 544 protocol.
  • AsyncLoopConsumer in abe/backends/message_queue/consumer.py is the default implementation that wraps a MessageQueueBackend and repeatedly invokes your handler coroutine.

EventConsumer protocol​

abe/backends/message_queue/base/consumer.py
class EventConsumer(Protocol):
    async def run(self, handler: Callable[[Dict[str, Any]], Awaitable[None]]) -> None: ...
    async def shutdown(self) -> None: ...

run(handler)​

  • Accepts an async handler that receives the message payload returned by the backend.
  • Implementations are expected to respect cancellation, propagate asyncio.CancelledError, and log other exceptions.

shutdown()​

  • Stops the consumer gracefully, ensuring in-flight messages complete where possible.
  • Should be idempotent: repeated calls after the consumer stops must be safe.

AsyncLoopConsumer​

abe/backends/message_queue/consumer.py
class AsyncLoopConsumer(EventConsumer):
    def __init__(self, backend: MessageQueueBackend, group: Optional[str] = None): ...
    async def run(self, handler: Callable[[Dict[str, Any]], Awaitable[None]]) -> None: ...
    async def shutdown(self) -> None: ...

Construction​

  • backend: The MessageQueueBackend instance returned by load_backend() or injected directly.
  • group (optional): Consumer group identifier forwarded to backend.consume() when supported.

Runtime behaviour​

  1. Creates an asyncio task that iterates over backend.consume(group=group).
  2. Awaits the provided handler for every payload. Exceptions are logged and do not terminate the loop unless configured otherwise.
  3. Tracks internal _running state and stores the current task for cancellation during shutdown.

Shutdown flow​

  • Cancels the running task if it is still active.
  • Handles asyncio.CancelledError and asyncio.TimeoutError explicitly, logging other exceptions at warning level.
  • Resets internal state (_task = None, _running = False).

Extending the consumer​

  • Subclass AsyncLoopConsumer to add retry/backoff logic or metrics.
  • Implement the EventConsumer protocol from scratch to integrate with orchestrators such as Celery or to bridge non-async backends.

Testing helpers​

  • Pass a fake backend that yields predetermined payloads from consume() to exercise handler logic.
  • Use pytest's asyncio support to run the consumer inside event loops and assert that shutdown() cancels tasks cleanly.