RabbitMQ

RabbitMQ is an AMQP message broker. @node-ts/bus-rabbitmq creates the exchanges and queues your handlers need, and retries failed messages after your retry strategy's delay, with no broker plugins. This page covers installing and configuring it.

@node-ts/bus-rabbitmqSource

Installation

npm

npm i @node-ts/bus-rabbitmq

pnpm

pnpm add @node-ts/bus-rabbitmq

yarn

yarn add @node-ts/bus-rabbitmq

Configure a RabbitMqTransport and pass it to the bus configuration:

import { Bus } from '@node-ts/bus-core'
import {
  RabbitMqTransport,
  RabbitMqTransportConfiguration
} from '@node-ts/bus-rabbitmq'
import { reserveRoomHandler } from './handlers/reserve-room-handler'
import { messageTypes } from './message-types.generated'

const rabbitConfiguration: RabbitMqTransportConfiguration = {
  queueName: 'reservations-service',
  deadLetterQueueName: 'reservations-service-dead-letter',
  connectionString: 'amqp://guest:guest@localhost',
  maxRetries: 5,
  // Survive a broker restart
  persistentMessages: true
}
const rabbitMqTransport = new RabbitMqTransport(rabbitConfiguration)

const bus = Bus.configure()
  .withMessageTypes(messageTypes)
  .withTransport(rabbitMqTransport)
  .withHandler(reserveRoomHandler)
  .build()

// Declares the exchanges and queues, and binds them for each handled message
await bus.initialize()
await bus.start()

Configuration

Option	Default	Description
queueName		The service queue to create and read messages from.
connectionString		An AMQP connection string, such as amqp://guest:guest@localhost.
deadLetterQueueName	dead-letter	Where messages go once they're out of attempts. Every service shares the default, so give each its own.
maxRetries	10	How many times a message is attempted before it goes to the dead letter queue.
persistentMessages	false	Whether messages survive a broker restart.
connectionRecovery	enabled	How to reconnect when the connection or channel is lost: exponential backoff from 100 ms to 30 s, retrying forever.

When the connection is lost, the transport reconnects, declares its topology again and carries on consuming. Messages that were being handled are redelivered by the broker.

Topology

For a service queue called <queue>, the transport declares:

• <queue>, the service queue. Each handled message's $name has a fanout exchange bound to it, as does each topicIdentifier of a custom handler.
• <queue>-retry-<n>ms, durable queues that hold returned messages until their retry delay expires, then put them at the back of the service queue. They're declared the first time they're needed, with <n> a power of two, so a short delay isn't stuck behind a long one. A message may wait up to twice its delay, but never less.
• <queue>-retry, a legacy retry exchange and queue that 1.x used. They're still declared, so messages already in them drain.
• the dead letter queue.

The number of failed attempts is kept in the failedAttempts message header.

Running RabbitMQ locally

docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management

See also

• Retry strategies
• RabbitMqTransportConfiguration in the API reference