@linagora/rabbitmq-client

An opinionated RabbitMQ client for Node.js that handles the boilerplate you'd otherwise copy-paste between services: confirm channels, dead letter queues, exponential backoff on publish, and automatic reconnection that restores your subscriptions.

Install

npm install @linagora/rabbitmq-client

Quick start

import { RabbitMQClient } from '@linagora/rabbitmq-client'

const client = new RabbitMQClient({ url: 'amqp://localhost' })
await client.init()

await client.publish('auth', 'user.created', { userId: '123' })

await client.subscribe('auth', 'user.created', 'my-service.user-created', async (msg) => {
  // your logic here
})

await client.close()

Configuration

Only url is required. Everything else has defaults.

| Option | Default | Description | |--------|---------|-------------| | url | -- | AMQP connection string | | maxRetries | 3 | How many times to retry a failed handler before sending the message to DLQ | | retryDelay | 1000 | Milliseconds between handler retries | | connectionRetryDelay | 5000 | Milliseconds between reconnection attempts | | initMaxAttempts | 5 | Connection attempts on startup before giving up | | publishMaxAttempts | 5 | Publish retries (exponential backoff, capped at 60s) | | prefetch | 10 | Channel prefetch: how many messages the broker delivers into consumer memory before waiting for acks | | concurrency | = prefetch | How many handlers run at once per subscription. Prefetch buffers messages in memory; concurrency bounds how many are processed simultaneously. Only throttles when set below prefetch (the broker never delivers more than prefetch unacked at a time); 0 means no limit. Override per queue via subscribe(..., { concurrency }) | | closeTimeout | 5000 | Milliseconds to wait for in-flight messages when closing | | logger | console (warn/error) | Any ILogger (console, Winston, pino, Bunyan). Omitted: logs warn/error to the console, stays quiet otherwise. Pass silentLogger for no output. | | hooks | -- | Observability callbacks (see Hooks) |

How it works

Publishing

Publishes go through a confirm channel, so you know the broker accepted the message. If something goes wrong, the client retries with exponential backoff and forces a new channel on each attempt to avoid retrying on a silently dead connection. Exchange assertions are cached per connection to avoid unnecessary AMQP round-trips on the hot path.

You can pass headers, correlation IDs, and other AMQP properties:

await client.publish('auth', 'user.created', { userId: '123' }, {
  headers: { 'x-trace-id': traceId },
  correlationId: requestId,
  messageId: uuid(),
  expiration: '60000', // TTL in ms
})

To reach one queue directly, publish to the default exchange ('') with the queue name as the routing key. The client does not declare the default exchange, since it always exists.

await client.publish('', 'orders.queue', { orderId: '42' })

A confirm only says the broker accepted the message, not that any queue got it. Pass mandatory: true to reject with UnroutableMessageError when no queue is bound to receive it. That error is not retried, since the same routing would fail the same way.

import { UnroutableMessageError } from '@linagora/rabbitmq-client'

try {
  await client.publish('', 'orders.queue', { orderId: '42' }, { mandatory: true })
} catch (error) {
  if (error instanceof UnroutableMessageError) {
    // orders.queue does not exist
  }
}

Subscribing

Each call to subscribe wires up the DLQ plumbing for you:

• A dead letter exchange (<exchange>.dlx)
• A dead letter queue (<queue>.dlq)
• A dead letter routing key (<routingKey>.dead)

The main queue is created as a quorum queue with at-least-once delivery and reject-publish overflow. Messages are manually acknowledged.

You can override the default queue arguments by passing a fifth argument:

await client.subscribe('events', 'order.placed', 'order-queue', handler, {
  queueArguments: { 'x-queue-type': 'classic', 'x-max-length': 10_000 },
})

Custom arguments are merged with the DLQ wiring defaults, so you can swap the queue type or add a max-length without losing dead-letter routing.

To limit how many messages this queue processes at once (for example, to protect a downstream database from a burst), set concurrency on the subscription. It defaults to the client-level concurrency, which in turn defaults to prefetch:

// Buffer up to 50 messages in memory, but process at most 4 at a time.
await client.subscribe('events', 'order.placed', 'order-queue', handler, {
  concurrency: 4,
})

One queue can take messages from several exchanges. Pass the other bindings in bindings; they share the queue, its consumer and its DLQ, and are restored on reconnection. Name the dead letter exchange after the queue with deadLetterExchange, so it does not live under another service's exchange. For exchanges another service owns, passiveExchanges checks they exist instead of declaring them, and subscribe fails when one is missing. Pass true for every exchange, or a list of names to check those and declare the rest:

await client.subscribe('space', 'twake.space.#', 'twake-space', handler, {
  bindings: [
    { exchange: 'b2b', routingKey: 'domain.user.deleted' },
    { exchange: 'auth', routingKey: 'user.deleted' },
  ],
  deadLetterExchange: 'twake-space.dlx',
  passiveExchanges: true,
})

To deliver every message to every instance of a service, subscribe with exclusive. Each instance gets its own queue, named <queue>.<uuid>, which the broker deletes when the connection closes or the consumer is cancelled. The client declares one under a new name on reconnection, and unsubscribe takes the queue you passed. A permission pattern that matches <queue>. followed by anything covers it, such as ^live(\..*)?$; an exact ^live$ does not. It has no DLQ: a message the handler fails is dropped, and so are messages published or in flight while the instance is disconnected. Nothing bounds it unless you set x-max-length.

await client.subscribe('live', '#', 'live', handler, {
  exclusive: true,
  queueArguments: { 'x-max-length': 10_000 },
})

Unsubscribing

Cancel a consumer and remove it from the auto-restoration list:

await client.unsubscribe('order-queue')

After unsubscribing, the queue will not be re-subscribed on reconnection.

Message handling

Incoming messages are JSON-parsed first. If that fails, the message goes straight to the DLQ (no point retrying garbage). Otherwise, your handler runs up to maxRetries times. Success means ack, final failure means nack to the DLQ.

A handler that knows retrying will not help throws DeadLetterError, and the message goes to the DLQ at once.

A subscription can set its own maxRetries, Infinity included, and maxRetryDelay, which doubles the delay between attempts from retryDelay up to that cap. With maxRetries: Infinity a message waits unacked while, say, the database is down, and keeps its place in the queue. close() and a reconnect stop the retries and leave the message unacked, so the broker redelivers it.

await client.subscribe('events', 'order.placed', 'order-queue', handler, {
  maxRetries: Infinity,
  maxRetryDelay: 60_000,
})

The handler also receives where the message came from, exchange and routingKey, and what the publisher and the broker attached to the delivery: headers, timestamp, messageId and correlationId. A handler that only needs the body can ignore it.

await client.subscribe('events', 'order.#', 'order-queue', async (message, { routingKey, headers }) => {
  // `x-death` is set by the broker on a message that went through a DLQ
  const replayed = headers['x-death'] !== undefined
})

Reconnection

If the connection or channel drops, the client reconnects and re-subscribes to everything automatically. Multiple reconnection triggers (e.g. connection close + channel close firing at the same time) are collapsed into a single attempt. A subscription that fails to restore is retried on its own, starting at connectionRetryDelay and doubling up to 60 seconds, until it succeeds, is unsubscribed, or the client closes. A consumer the broker cancels, for example because its queue was deleted, is re-subscribed the same way.

Graceful shutdown

close() waits for in-flight message handlers to finish before tearing down the channel, up to closeTimeout milliseconds. If handlers don't drain in time, the client closes anyway and logs a warning.

// wait for handlers to finish, then close
await client.close()

// close but keep subscriptions for a later init()
await client.close(false)

Health check

checkHealth() creates and immediately deletes a temporary queue. Useful for Kubernetes readiness probes.

const healthy = await client.checkHealth()

Hooks

Optional callbacks for wiring metrics, tracing, or alerting. Hook errors are swallowed so they never break message flow.

const client = new RabbitMQClient({
  url: 'amqp://localhost',
  hooks: {
    onPublish({ exchange, routingKey, attempts }) {
      metrics.increment('rabbitmq.publish', { exchange })
    },
    onMessageProcessed({ exchange, routingKey, duration, attempts }) {
      metrics.histogram('rabbitmq.handler.duration', duration)
    },
    onMessageDlq({ exchange, routingKey, duration, reason }) {
      alerting.warn(`Message sent to DLQ: ${reason}`)
    },
    onReconnect({ subscriptionsRestored, subscriptionsFailed }) {
      metrics.increment('rabbitmq.reconnect')
    },
  },
})

| Hook | Fires when | |------|-----------| | onPublish | A message is confirmed by the broker | | onMessageProcessed | A handler completes successfully (includes duration and retry count) | | onMessageDlq | A message is nacked to the DLQ, with reason 'invalid_json', 'max_retries_exhausted' or 'dead_letter_error' | | onReconnect | The client reconnects and re-establishes subscriptions |

Test helpers

The @linagora/rabbitmq-client/testing entrypoint provides mocks that don't depend on any test framework.

import { createMockAmqplib } from '@linagora/rabbitmq-client/testing'

const { mockConnection, amqpMock } = createMockAmqplib()

// check what was published
mockConnection.channel.getPublishedMessages()

// feed a message into a consumer
mockConnection.channel.simulateMessage({ userId: '123' })

// simulate a broker going down
mockConnection.simulateClose()

License

MIT