Event Bus Implementations: Redis, Postgres, RabbitMQ

HexCore ships four event bus implementations that all satisfy the IEventBus / AbstractEventBus contract. Every bus exposes the same two-method interface (subscribe + publish) and supports Smart Routing — if an event handler carries the __cqrs_background_handler__ attribute (set by @background_handler), the bus automatically delegates execution to the configured ITaskEnqueuer instead of running the handler inline.

Shared interface

class IEventBus(abc.ABC):
    def subscribe(self, event_type: type[DomainEvent], handler: Callable) -> None: ...
    async def publish(self, event: DomainEvent) -> None: ...

Distributed buses additionally expose:

async def start_consuming(self) -> None: ...
async def stop(self) -> None: ...

Implementations

InMemoryEventBus
RedisEventBus
PostgresEventBus
RabbitMQEventBus

hexcore.application.cqrs.in_memory_buses.InMemoryEventBusPure in-process event bus with Smart Routing support. Handlers execute in the same asyncio event loop. Background handlers are delegated to the configured ITaskEnqueuer. This bus is the default value of ServerConfig.event_bus.

Constructor

InMemoryEventBus(
    pipeline: MiddlewarePipeline | None = None,
    enqueuer: ITaskEnqueuer | None = None,
    serializer: ISerializer | None = None,
)

pipeline

MiddlewarePipeline | None

default:"None"

Optional middleware pipeline that wraps every handler invocation. Defaults to an empty MiddlewarePipeline() when None.

enqueuer

ITaskEnqueuer | None

default:"None"

Task enqueuer required when any subscribed handler is decorated with @background_handler. If None and a background handler is called, a RuntimeError is raised.

serializer

ISerializer | None

default:"None"

Serialiser required when enqueuer is provided. Used to convert domain events to a wire-safe dict before enqueuing.

Methods

Appends handler to the list of callables subscribed to event_type. Multiple handlers for the same event type are all called in registration order.

publish(event: DomainEvent)

Awaitable[None]

Dispatches event to every registered handler. Handlers with __cqrs_background_handler__ = True are routed to enqueuer.enqueue_handler(handler_name, payload, queue). All others are awaited inline through the middleware pipeline.

Smart Routing logic

for event_handler in handlers:
    is_background = getattr(event_handler, "__cqrs_background_handler__", False)

    if is_background:
        handler_name = getattr(event_handler, "__cqrs_handler_name__")
        queue_name   = getattr(event_handler, "__cqrs_queue__", "default")
        payload      = self._serializer.serialize(event)
        await self._enqueuer.enqueue_handler(handler_name, payload, queue=queue_name)
    else:
        await event_handler(event)  # via pipeline

Example

from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
from myapp.celery_app import celery_enqueuer

bus = InMemoryEventBus(
    enqueuer=celery_enqueuer,
    serializer=PydanticSerializer(),
)

bus.subscribe(UserCreatedEvent, on_user_created)

await bus.publish(UserCreatedEvent(user_id=user.id))

hexcore.infrastructure.cqrs.redis_bus.RedisEventBusDurable, ordered event bus backed by Redis Streams with Consumer Group support. Messages are persisted in the stream until explicitly acknowledged, making this suitable for reliable multi-consumer fan-out.

Constructor

RedisEventBus(
    redis_client: "Redis",
    serializer: "PydanticSerializer",
    stream_name: str,
    group_name: str,
    consumer_name: str | None = None,
    enqueuer: ITaskEnqueuer | None = None,
)

redis_client

redis.asyncio.Redis

required

An active redis.asyncio.Redis instance. Manage its lifecycle (connect/close) in your application lifespan.

serializer

PydanticSerializer

required

Serialiser used to convert domain events to JSON-serialisable dicts before writing to the stream, and to reconstruct events when consuming.

stream_name

str

required

Redis stream key (e.g. "hexcore:events"). Created automatically with MKSTREAM on the first start_consuming() call.

group_name

str

required

Consumer group name. Multiple workers sharing the same group compete for messages (work-queue fan-out). Each microservice should use a distinct group to receive all events independently.

consumer_name

str | None

default:"None"

Unique name for this consumer instance within the group. Defaults to "consumer-{uuid4()}" when None, ensuring multiple pods of the same service don’t collide.

enqueuer

ITaskEnqueuer | None

default:"None"

Optional enqueuer for Smart Routing background handlers.

Methods

Registers a handler locally and records the event type’s fully-qualified name (module.qualname) for deserialization lookup. Call before start_consuming().

publish(event: DomainEvent)

Awaitable[None]

Serialises the event and appends it to the Redis stream via XADD stream_name * payload <json>.

start_consuming()

Awaitable[None]

Enters a XREADGROUP polling loop (blocking up to 1000 ms per iteration, reading up to 10 messages at a time). Dispatches each message to registered handlers and issues XACK on success. Runs until stop() is called.

stop()

Awaitable[None]

Sets the internal stop event, causing the start_consuming() loop to exit gracefully after the current batch.

Example

import redis.asyncio as aioredis
from hexcore.infrastructure.cqrs.redis_bus import RedisEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer

redis_client = aioredis.Redis.from_url("redis://localhost:6379/0")

bus = RedisEventBus(
    redis_client=redis_client,
    serializer=PydanticSerializer(),
    stream_name="hexcore:events",
    group_name="notification-service",
)

bus.subscribe(OrderShippedEvent, send_shipping_email)

# In a background task / worker process:
await bus.start_consuming()

hexcore.infrastructure.cqrs.postgres_bus.PostgresEventBusLightweight event bus that uses PostgreSQL LISTEN / NOTIFY. Ideal when you already run PostgreSQL and want pub/sub without adding Redis or RabbitMQ to your stack.

NOTIFY does not persist events. If no consumer is listening at the moment publish() is called, the event is silently dropped. This bus is suitable for ephemeral notifications (cache invalidation, UI push), not for guaranteed delivery or work queues.

Constructor

PostgresEventBus(
    pool: "asyncpg.Pool",
    serializer: "PydanticSerializer",
    channel_name: str,
    enqueuer: ITaskEnqueuer | None = None,
)

pool

asyncpg.Pool

required

An active asyncpg connection pool. The bus acquires a dedicated listener connection from this pool when start_consuming() is called.

serializer

PydanticSerializer

required

Serialiser used to convert events to JSON for pg_notify and to reconstruct them on receipt.

channel_name

str

required

PostgreSQL notification channel name (e.g. "hexcore_events"). All publishers and subscribers must use the same channel.

enqueuer

ITaskEnqueuer | None

default:"None"

Optional enqueuer for Smart Routing background handlers.

Methods

Registers a handler locally and records the FQN for deserialization lookup.

publish(event: DomainEvent)

Awaitable[None]

Acquires a connection from the pool and executes SELECT pg_notify($1, $2) with the channel name and the JSON-serialised event payload.

start_consuming()

Awaitable[None]

Acquires a dedicated persistent connection, registers an asyncpg listener via add_listener(channel_name, callback), and then awaits a stop signal. Each NOTIFY callback creates an asyncio.Task to process the message without blocking the listener connection.

stop()

Awaitable[None]

Sets the internal stop event. Removes the listener and releases the dedicated connection back to the pool.

Example

import asyncpg
from hexcore.infrastructure.cqrs.postgres_bus import PostgresEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer

pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")

bus = PostgresEventBus(
    pool=pool,
    serializer=PydanticSerializer(),
    channel_name="hexcore_events",
)

bus.subscribe(UserCreatedEvent, on_user_created)

await bus.start_consuming()  # blocks until stop() is called

hexcore.infrastructure.cqrs.rabbitmq.RabbitMQEventBusFull-featured distributed event bus over RabbitMQ using aio-pika. Uses a TOPIC exchange for flexible routing-key-based fan-out. Each service declares its own durable queue and binds routing keys for the events it cares about.

Constructor

RabbitMQEventBus(
    connection: "AbstractRobustConnection",
    serializer: AbstractSerializer,
    pipeline: MiddlewarePipeline | None = None,
    exchange_name: str = "hexcore.events",
    queue_name: str = "hexcore.worker.queue",
)

connection

aio_pika.abc.AbstractRobustConnection

required

A robust aio-pika connection. Use aio_pika.connect_robust(url) and manage its lifecycle in your application startup/shutdown hooks.

serializer

AbstractSerializer

required

Serialiser implementation (e.g. PydanticSerializer). Used for both publish and consume paths.

pipeline

MiddlewarePipeline | None

default:"None"

Optional middleware pipeline wrapping every handler invocation during consumption. Defaults to an empty MiddlewarePipeline([]).

exchange_name

str

default:"\"hexcore.events\""

Name of the durable TOPIC exchange. Declared lazily on first publish() or start_consuming() call.

queue_name

str

default:"\"hexcore.worker.queue\""

Name of the durable queue exclusive to this service/worker. Multiple pods of the same service should share a queue name to achieve work-queue semantics.

Methods

Registers the handler locally. The routing key is derived from the event class name: event_type.__name__.replace("Event", "").upper(). Queue binding happens at start_consuming().

publish(event: DomainEvent)

Awaitable[None]

Calls _setup() lazily (declares exchange if needed), serialises the event, and publishes a persistent aio_pika.Message to the exchange with routing_key = event.event_name.

start_consuming()

Awaitable[None]

Declares the durable queue, binds a routing key for every subscribed event type, then calls queue.consume(callback) to start receiving messages. Messages are processed concurrently via asyncio.gather. Failed messages are NACK-ed and requeued.

Example

import aio_pika
from hexcore.infrastructure.cqrs.rabbitmq import RabbitMQEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer

connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")

bus = RabbitMQEventBus(
    connection=connection,
    serializer=PydanticSerializer(),
    exchange_name="myapp.events",
    queue_name="notification-worker",
)

bus.subscribe(OrderPlacedEvent, handle_order_placed)

await bus.start_consuming()  # never returns; run in background task

`ILockProvider`

hexcore.domain.cqrs.cron.ILockProvider Abstract distributed lock interface used by DynamicScheduler to prevent duplicate cron job execution across multiple replicas.

class ILockProvider(abc.ABC):
    @abc.abstractmethod
    async def acquire_lock(self, lock_key: str, ttl_seconds: int) -> bool: ...

    @abc.abstractmethod
    async def release_lock(self, lock_key: str) -> None: ...

acquire_lock(lock_key: str, ttl_seconds: int)

Awaitable[bool]

Attempts to acquire the lock identified by lock_key. Returns True if the lock was acquired (no other holder), False if already locked. The lock expires automatically after ttl_seconds to prevent deadlocks.

release_lock(lock_key: str)

Awaitable[None]

Explicitly releases the lock before its TTL expires. Optional if TTL-based expiry is acceptable.

`RedisLockProvider`

hexcore.infrastructure.cqrs.redis_lock.RedisLockProvider Implements ILockProvider using Redis SET key value NX EX ttl for atomic, TTL-enforced mutual exclusion.

Constructor

RedisLockProvider(redis_client: "redis.asyncio.Redis")

redis_client

redis.asyncio.Redis

required

An active redis.asyncio.Redis instance.

import redis.asyncio as aioredis
from hexcore.infrastructure.cqrs.redis_lock import RedisLockProvider

redis_client = aioredis.Redis.from_url("redis://localhost:6379/0")
lock_provider = RedisLockProvider(redis_client=redis_client)

acquired = await lock_provider.acquire_lock("cron:daily_report:2024-01-15T02:00:00", ttl_seconds=300)
if acquired:
    # run the job
    await lock_provider.release_lock("cron:daily_report:2024-01-15T02:00:00")

`PostgresLockProvider`

hexcore.infrastructure.cqrs.postgres_lock.PostgresLockProvider Implements ILockProvider using a dedicated PostgreSQL table (hexcore_cron_locks) with UPSERT-based conditional locking.

Constructor

PostgresLockProvider(
    pool: "asyncpg.Pool | asyncpg.Connection",
    table_name: str = "hexcore_cron_locks",
)

pool

asyncpg.Pool | asyncpg.Connection

required

An active asyncpg pool or single connection.

table_name

str

default:"\"hexcore_cron_locks\""

Name of the lock table. Customise when sharing a schema with strict naming conventions.

`setup()`

async def setup(self) -> None

Creates the hexcore_cron_locks table if it does not exist. Must be called once at application startup before any lock is acquired.

import asyncpg
from hexcore.infrastructure.cqrs.postgres_lock import PostgresLockProvider

pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")
lock_provider = PostgresLockProvider(pool=pool)
await lock_provider.setup()   # create table if needed

acquired = await lock_provider.acquire_lock("cron:cleanup:2024-01-15T03:00:00", ttl_seconds=120)

PostgresLockProvider uses an INSERT ... ON CONFLICT DO UPDATE ... WHERE expires_at < NOW() query. This means an expired lock is automatically taken over by the first new acquirer without requiring an explicit release_lock().

Domain

CQRS

Infrastructure

Event Bus Implementations: Redis, Postgres, RabbitMQ

Shared interface

Implementations

Constructor

Methods

Smart Routing logic

Example

Constructor

Methods

Example

Constructor

Methods

Example

Constructor

Methods

Example

`ILockProvider`

`RedisLockProvider`

Constructor

`PostgresLockProvider`

Constructor

`setup()`

Build docs developers (and LLMs) love

Domain

CQRS

Infrastructure

Documentation Index

​Shared interface

​Implementations

​Constructor

​Methods

​Smart Routing logic

​Example

​Constructor

​Methods

​Example

​Constructor

​Methods

​Example

​Constructor

​Methods

​Example

​ILockProvider

​RedisLockProvider

​Constructor

​PostgresLockProvider

​Constructor

​setup()

Build docs developers (and LLMs) love

Shared interface

Implementations

Constructor

Methods

Smart Routing logic

Example

Constructor

Methods

Example

Constructor

Methods

Example

Constructor

Methods

Example

`ILockProvider`

`RedisLockProvider`

Constructor

`PostgresLockProvider`

Constructor

`setup()`