amqplib Documentation

Automatic recovery can be enabled by passing a recovery object in the options of amqplib.connect(). This allows the client to automatically attempt to reconnect and recreate the topology (queues, exchanges, consumers) after a connection loss.

Key options for the recovery object:

• initialDelay: Initial delay in ms before first retry.
• maxDelay: Maximum delay in ms between retries.
• factor: Multiplier for the delay between retries.
• jitter: Randomness factor applied to the delay.
• maxRetries: Maximum number of reconnection attempts (Infinity for unlimited).
• async setup(model): An async function called after every successful (re)connect. Use this to recreate your topology (e.g., assertQueue, consume).

const amqplib = require('amqplib');

const connection = await amqplib.connect('amqp://localhost', {
  recovery: {
    initialDelay: 200, // ms
    maxDelay: 5000, // ms
    factor: 2,
    jitter: 0.2,
    maxRetries: Infinity,
    async setup(model) {
      // Called after every successful (re)connect.
      // Recreate topology/consumers here.
      const ch = await model.createChannel();
      await ch.assertQueue('tasks', {durable: true});
    },
  },
});

connection.on('connect', () => {
  console.log('connected');
});

connection.on('disconnect', (err) => {
  console.warn('disconnected', err.message);
});

If a synchronous error is thrown inside one of your event listeners (e.g., conn.on('message', ...)), it can cause the connection or channel to close or be swallowed.

To prevent this, register a handler-error listener on both the connection and each channel. The handler-error event is emitted when your own event listener throws an error. This is distinct from the error event, which is emitted by amqplib for protocol-level errors.

• error event: Emitted by amqplib for protocol errors.
• handler-error event: Emitted when a user-supplied handler throws. Receives (err, event) where event is the name of the event whose handler threw.

const connection = await amqp.connect('amqp://localhost');

connection.on('error', (err) => { /* handle protocol errors */ });
connection.on('handler-error', (err, event) => {
  console.error(`Uncaught exception in connection ${event} listener:`, err);
});

const channel = await connection.createChannel();

channel.on('error', (err) => { /* handle protocol errors */ });
channel.on('handler-error', (err, event) => {
  console.error(`Uncaught exception in channel ${event} listener:`, err);
});

The Promise/Async API is the standard way to use amqplib with modern async/await syntax. It provides methods for connecting to an AMQP server, creating channels, asserting queues, consuming messages, and sending messages to queues.

const amqplib = require('amqplib');

(async () => {
  const queue = 'tasks';
  const conn = await amqplib.connect('amqp://localhost');
  conn.on('error', (err) => { console.error('Connection error:', err); });
  conn.on('handler-error', (err, event) => { console.error(`Uncaught exception in connection ${event} listener:`, err); });

  const ch1 = await conn.createChannel();
  ch1.on('error', (err) => { console.error('Channel error:', err); });
  ch1.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
  await ch1.assertQueue(queue);

  // Listener
  ch1.consume(queue, (msg) => {
    if (msg !== null) {
      console.log('Received:', msg.content.toString());
      ch1.ack(msg);
    } else {
      console.log('Consumer cancelled by server');
    }
  });

  // Sender
  const ch2 = await conn.createChannel();
  ch2.on('error', (err) => { console.error('Channel error:', err); });
  ch2.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });

  setInterval(() => {
    ch2.sendToQueue(queue, Buffer.from('something to do'));
  }, 1000);
})();

For legacy codebases or specific requirements, amqplib provides a callback-based API via the amqplib/callback_api module.

const amqplib = require('amqplib/callback_api');
const queue = 'tasks';

amqplib.connect('amqp://localhost', (err, conn) => {
  if (err) throw err;

  conn.on('error', (err) => { console.error('Connection error:', err); });
  conn.on('handler-error', (err, event) => { console.error(`Uncaught exception in connection ${event} listener:`, err); });

  // Listener
  conn.createChannel((err, ch2) => {
    if (err) throw err;

    ch2.on('error', (err) => { console.error('Channel error:', err); });
    ch2.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });

    ch2.assertQueue(queue);

    ch2.consume(queue, (msg) => {
      if (msg !== null) {
        console.log(msg.content.toString());
        ch2.ack(msg);
      } else {
        console.log('Consumer cancelled by server');
      }
    });
  });

  // Sender
  conn.createChannel((err, ch1) => {
    if (err) throw err;

    ch1.on('error', (err) => { console.error('Channel error:', err); });
    ch1.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
    ch1.assertQueue(queue);

    setInterval(() => {
      ch1.sendToQueue(queue, Buffer.from('something to do'));
    }, 1000);
  });
});