rabbitmq-client

The RPCClient allows for request-response patterns between services using RabbitMQ.

• RPC Server: Create a Consumer on a specific queue. The callback receives the request (req) and a reply function. Calling await reply(data) sends the response back.
• RPC Client: Create an RPCClient and use .send(queue, message) to send a request and wait for the response.

// rpc-server.js
const rpcServer = rabbit.createConsumer({
  queue: 'my-rpc-queue'
}, async (req, reply) => {
  console.log('request:', req.body)
  await reply('pong')
})

// rpc-client.js
const rpcClient = rabbit.createRPCClient({confirm: true})
const res = await rpcClient.send('my-rpc-queue', 'ping')
console.log('response:', res.body) // pong

To use rabbitmq-client, initialize a Connection with an AMQP URI. The library provides high-level Consumer and Publisher interfaces that handle automatic reconnection, re-subscription, and retries.

Key lifecycle events for the Connection include error and connection (emitted when a connection is successfully established or re-established).

import {Connection} from 'rabbitmq-client'

// Initialize:
const rabbit = new Connection('amqp://guest:guest@localhost:5672')
rabbit.on('error', (err) => {
  console.log('RabbitMQ connection error', err)
})
rabbit.on('connection', () => {
  console.log('Connection successfully (re)established')
})

The ConsumerHandler can return a ConsumerStatus to explicitly control how RabbitMQ acknowledges or rejects a message. This is useful for fine-grained control over message lifecycle.

import { ConsumerStatus } from 'rabbitmq-client'

// Inside your handler:
if (isTransientError) {
  return ConsumerStatus.REQUEUE
} else if (isFatalError) {
  return ConsumerStatus.DROP
}

The library uses specific interfaces to represent messages received from RabbitMQ:

AsyncMessage

Received via a consumer (e.g., Channel.basicConsume). It includes:

• body: The actual message content (string, Buffer, or any).
• deliveryTag: The server-assigned delivery tag.
• exchange: The exchange the message was published to.
• routingKey: The routing key used.
• redelivered: Boolean indicating if the message was previously delivered.
• HeaderFields: Metadata like contentType, correlationId, etc.

SyncMessage

Received via a synchronous pull (e.g., Channel.basicGet). It includes the same fields as AsyncMessage plus:

• messageCount: The number of messages remaining in the queue.

When using a connection string (url), you can pass specific RabbitMQ configuration parameters as query parameters. These will override the default values or the values provided in the object.

Supported URL parameters:

• heartbeat: Sets the heartbeat interval in seconds.
• connection_timeout: Sets the connection timeout in milliseconds.
• channel_max: Sets the maximum number of channels.

Note: The heartbeat value must be $\ge 0$. The channel_max value is clamped between $1$ and the value provided in maxChannels.

// Example using URL parameters
const url = 'amqp://guest:guest@localhost:5672/?heartbeat=30&connection_timeout=5000&channel_max=100';

If the high-level APIs are insufficient, you can acquire a raw AMQP Channel using rabbit.acquire().

Warning: When using raw channels, you are responsible for:

1. Handling connection/channel loss (re-declaring queues, exchanges, etc.).
2. Closing the channel manually using await ch.close() to prevent resource leaks.

Raw channels can emit events like close.

// Will wait for the connection to establish and then create a Channel
const ch = await rabbit.acquire()

ch.on('close', () => {
  console.log('channel was closed')
})

const msg = ch.basicGet('my-queue')
console.log(msg)

// It's your responsibility to close any acquired channels
await ch.close()

While it is recommended to use the declarative Consumer or Publisher APIs for managing infrastructure, you can use the Connection interface for one-off management tasks:

• queueDeclare({queue, exclusive, passive, ...})
• exchangeDeclare({queue, exchange, type, ...})
• queueBind({queue, exchange, ...})

Note: queueDeclare with passive: true returns an object containing the messageCount.

const rabbit = new Connection()

await rabbit.queueDeclare({queue: 'my-queue', exclusive: true})

await rabbit.exchangeDeclare({queue: 'my-queue', exchange: 'my-exchange', type: 'topic'})

await rabbit.queueBind({queue: 'my-queue', exchange: 'my-exchange'})

const {messageCount} = await rabbit.queueDeclare({queue: 'my-queue', passive: true})

The Publisher API provides a reliable way to send messages. It supports publish confirmations and automatic retries.

Configuration options:

• confirm: Set to true to enable publish confirmations.
• maxAttempts: Number of retries for publishing.
• exchanges: An array of exchanges to ensure exist.

You can publish to a specific exchange using a routing key, or publish directly to a queue name.

const pub = rabbit.createPublisher({
  confirm: true,
  maxAttempts: 2,
  exchanges: [{exchange: 'my-events', type: 'topic'}]
})

// Publish to an exchange with a routing key
await pub.send(
  {exchange: 'my-events', routingKey: 'users.visit'},
  {id: 1, name: 'Alan Turing'}
)

// Or publish directly to a queue
await pub.send('user-events', {id: 1, name: 'Alan Turing'})