RabbitMQ Command and Event Buses

RabbitMQ support for node-cqrs provides:

• a command bus for distributing commands across application instances;
• an event bus for broadcasting events or creating durable worker queues;
• automatic connection recovery and subscription restoration;
• publisher confirms, consumer acknowledgements, dead-letter queues, and OpenTelemetry context propagation.

Installation

Install the RabbitMQ client alongside node-cqrs:

npm install node-cqrs amqplib

Choose what you need

All exchanges are durable topic exchanges. Message types are routing keys, so subscriptions can select a type without receiving unrelated messages.

Container setup

Register a connection factory, one shared gateway, and the buses the application needs:

import * as amqplib from 'amqplib';
import { ContainerBuilder, type IContainer } from 'node-cqrs';
import { RabbitMqCommandBus, RabbitMqEventBus, RabbitMqGateway } from 'node-cqrs/rabbitmq';

type CredentialsStore = {
	getRabbitMqUrl(): Promise<string> | string;
};

interface AppContainer extends IContainer {
	credentialsStore: CredentialsStore;
}

const builder = new ContainerBuilder<AppContainer>();

builder.register(container => async () => {
	const url = await container.credentialsStore.getRabbitMqUrl();
	return amqplib.connect(url);
}, 'rabbitMqConnectionFactory');

builder.register(RabbitMqGateway).as('rabbitMqGateway');
builder.register(RabbitMqCommandBus).as('commandBus');
builder.register(RabbitMqEventBus).as('eventBus');

The factory must create a new connection each time it is called. RabbitMqGateway owns that connection, opens it lazily, closes it during shutdown, and calls the factory again when reconnecting. It also restores recorded subscriptions after reconnecting.

For static configuration, the factory can connect directly:

builder.register(() => () => amqplib.connect(process.env.RABBITMQ_URL ?? 'amqp://localhost'),
	'rabbitMqConnectionFactory');

Commands

RabbitMqCommandBus publishes commands to a durable queue. When several application instances consume the same queue, RabbitMQ delivers each command to one of them.

await commandBus.on('sendWelcomeEmail', async command => {
	await sendWelcomeEmail(command.aggregateId!, command.payload!.email);
});

await commandBus.send('sendWelcomeEmail', userId, {
	payload: { email: '[email protected]' }
});

The default exchange is node-cqrs.commands and the default queue is node-cqrs.commands.default. Use a different durable queue for each independently deployed command-handling service:

builder.registerInstance({
	exchange: 'application.commands',
	queueName: 'accounts.commands',
	concurrentLimit: 20
}, 'rabbitMqCommandBusConfig');

send() resolves after RabbitMQ confirms the publish. It does not wait for command execution or return the handler’s result.

Events

RabbitMqEventBus uses a temporary exclusive queue by default. Every connected gateway gets its own queue, so each running application instance receives each matching event. The queue is deleted on disconnect, and events published while that instance is disconnected are not retained for it.

await eventBus.on('userCreated', async event => {
	await notifyUser(event.aggregateId!);
});

await eventBus.publish({
	type: 'userCreated',
	aggregateId: userId
});

The publishing gateway does not receive its own events by default. Set ignoreOwn: false when the same process must handle events it publishes:

builder.registerInstance({
	exchange: 'application.events',
	ignoreOwn: false
}, 'rabbitMqEventBusConfig');

Subscribe to every event type with RabbitMqEventBus.allEventsWildcard.

Durable worker queues

Use a named queue when an event must be retained during worker downtime and processed by only one worker in a group. Calls using the same queue name share the work across application instances:

const emailQueue = eventBus.queue('welcome-email');

await emailQueue.on('userCreated', async event => {
	await sendWelcomeEmail(event.aggregateId!, event.payload!.email);
});

registerSaga() uses this queue-provider API to give each saga type its own durable event queue.

Alternatively, set queueName in rabbitMqEventBusConfig when one active instance should receive the event stream and other instances should remain on standby. These subscriptions enable RabbitMQ’s single-active-consumer mode.

Event bridge for a shared event store

When application instances share one event store, each instance publishes the events it commits to its own event bus, so its in-memory views and receptors do not see events committed by other instances. RabbitMqEventBridge delivers them through RabbitMQ when the event store cannot notify instances about new commits itself; the PostgreSQL event storage can, and needs no bridge.

Keep the local in-memory event bus and register one bridge instance in two roles: as an event feed, and as the last processor of the local dispatch pipeline:

builder.register(RabbitMqGateway).as('rabbitMqGateway');
builder.registerEventFeed(RabbitMqEventBridge).as('rabbitMqEventBridge');
builder.register(c => [...c.defaultEventDispatchPipeline, c.rabbitMqEventBridge!]).as('eventDispatchPipeline');

• In the dispatch pipeline, after event storage, it publishes each event committed by its instance to the node-cqrs.events exchange.
• As a feed, it consumes events published by other instances and dispatches them locally with meta.origin set to the feed origin, 'external' by default, through the feed pipeline, which is empty by default.

The bridge awaits the broker’s confirmation of each publish before events reach the local event bus, so every dispatch includes a RabbitMQ round trip. By default, a failed publish is reported to eventPublishErrorHandler, or logged, and the dispatch continues: local projections receive the event, other instances do not. To fail the dispatch instead, so that the event is not distributed further, set throwOnPublishError:

builder.registerInstance({ throwOnPublishError: true }, 'rabbitMqEventBridgeConfig');

The dispatch then rejects with the publish error and the event does not reach the local event bus. Whether the event remains stored depends on the event storage. A custom eventPublishErrorHandler that throws has the same effect on the bridge, but the dispatcher calls the same handler for failed event bus publishes, where a thrown error is not handled; prefer throwOnPublishError.

An event committed by an instance that stops before publishing it is not delivered to other instances; their projections pick it up the next time they restore. Events published while an instance is disconnected are lost for it unless the bridge uses a durable queue.

A message is acknowledged once its dispatch completes and rejected when a feed pipeline processor fails. Use the bridge instead of RabbitMqEventBus, not alongside it on the same exchange: an instance using RabbitMqEventBus as its event bus already receives every event, and would receive events of other instances twice. The bridge publishes only events of the EventStore.DEFAULT_ORIGIN origin, or without an origin, and fails the dispatch of any other, so it cannot be placed in a feed pipeline by mistake.

Delivery and failures

Matching handlers of a message run concurrently, so a handler awaiting another one’s outcome, such as a receptor waiting for a projection, does not block it. The message is acknowledged after all of them complete successfully; when any of them fails, the others still complete and the message is rejected with the failure, or with an AggregateError when several handlers fail. RabbitMQ can redeliver a message when a connection closes before its acknowledgement, including when the handler already produced external effects. Handlers should therefore be idempotent or deduplicate messages by event id.

When a handler throws or exceeds handlerProcessTimeout, the message is rejected without requeueing:

• command queues create ${queueName}.failed by default;
• event bus dead-lettering is disabled by default, including for named event queues;
• when dead-lettering is disabled, rejected messages are discarded.

Enable and monitor dead-lettering for durable event processing:

builder.registerInstance({
	queueName: 'accounts.events',
	deadLetterQueue: true,
	handlerProcessTimeout: 30_000
}, 'rabbitMqEventBusConfig');

The failed queue is bound through the ${exchange}.failed exchange. The adapter does not automatically retry failed messages or consume the failed queue; recovery policy belongs to the application.

Configuration

Both bus configurations may be an object, a synchronous factory, or an asynchronous factory.

Changing queue arguments for an existing queue causes RabbitMQ to reject the declaration. Delete and recreate the queue, or migrate to a new queue name, when changing options such as dead-lettering, TTL, or expiry.

Application identity

rabbitMqAppId identifies the publishing gateway for ignoreOwn filtering. A random id is generated by default. If an explicit provider is registered, it should return an id unique to the running process, not one shared by the whole deployment:

builder.registerInstance(process.env.INSTANCE_ID!, 'rabbitMqAppId');

Register a provider function instead when the instance id must be resolved lazily or asynchronously.

Operations

• Use unique queue names for independently processed workloads. Consumers sharing a queue compete for the same messages rather than each receiving a copy.
• Named durable queues use RabbitMQ quorum queues. Queue expiry on quorum queues requires RabbitMQ 3.10 or later.
• Monitor connection lifecycle events, failed queues, queue depth, consumer count, and processing latency.
• Set concurrentLimit according to handler cost and downstream capacity. Unlimited prefetch can place many unacknowledged messages on one process.
• Set handlerProcessTimeout above the longest valid handler duration. A timed-out handler continues running in JavaScript even though its message has already been rejected.
• Shut down through the application’s normal SIGINT or SIGTERM path, or call await rabbitMqGateway.disconnect() explicitly.

rabbitMqGateway.on('connected', () => health.setRabbitMqReady(true));
rabbitMqGateway.on('disconnected', () => health.setRabbitMqReady(false));

Advanced APIs

Most consumers only need the command and event buses. Use RabbitMqGateway directly for custom exchanges, subscription options, or lifecycle monitoring. Its publish(), subscribe(), and unsubscribe() methods expose the lower-level transport while retaining connection recovery, acknowledgements, dead-lettering, and tracing.

Run locally

Start the repository’s RabbitMQ test service:

docker compose -f tests/integration/rabbitmq/docker-compose.yml up -d

Run the integration tests:

npm run test:rabbitmq

Stop RabbitMQ when finished:

docker compose -f tests/integration/rabbitmq/docker-compose.yml down