amqp-connection-manager: reconnecting and round robin channels over amqplib

The problem is that a dead broker takes the client with it

The project describes itself as connection management for amqplib, a wrapper around amqplib which provides automatic reconnects. The README states the problem in a memorable way: the library automatically reconnects when your amqplib broker dies in a fire. That is the entire pitch, and it is narrower than it first appears. amqplib gives you a channel and a promise-based API, and it assumes the connection stays up. In production the broker does not cooperate, so a Node service that holds a long-lived channel dies with it or hangs on it. The audience is a backend engineer running a consumer or publisher against RabbitMQ who has been woken by a pager, and the design constraint is that the fix has to be transparent enough that the rest of the application code barely changes. The README describes the library as very un-opinionated, a thin wrapper around amqplib, which is both the selling point and the limitation.

Install it alongside amqplib, which stays a peer dependency

Installation is one npm command, and it installs both packages:

npm install --save amqplib amqp-connection-manager

The package metadata makes the relationship explicit. `amqplib` is a peer dependency with the wildcard range `*`, and the three direct dependencies are `promise-breaker`, `typed-emitter`, and a chai promise typing package used for the tests. The `promise-breaker` dependency is what delivers the dual-style API, since the README says the library supports both promises and callbacks using promise-breaker, and notes that much like amqplib, any function which returns a Promise will also accept a callback as an optional parameter. The package is version 5.0.1, requires Node 22 or newer, and ships dual ESM and CommonJS builds with separate type declaration output, so a project on either module system can consume it. The licence on the repository is reported as unknown, so the terms need checking in the package files themselves.

A ChannelWrapper is a proxy that may or may not have a channel

The core abstraction is the ChannelWrapper. The README describes it as a proxy for the actual channel, which may or may not exist at any moment depending on whether you are currently connected. That single sentence is the design. Your code holds a stable object and calls `sendToQueue` or `publish` on it, and whether that reaches a broker, sits in a buffer, or fails, is not your problem. The basics section explains the intended workflow: you usually do some setup work when you create a channel, such as asserting that queues or exchanges exist or binding to queues, and then you send and receive messages and never touch that setup again. The manager reconnects to a new broker whenever the current one dies, and when you ask for a channel you specify one or more `setup` functions which are run every time it reconnects, to make sure your channel and broker are in a sane state. The setup function is the piece that makes reconnection safe rather than merely possible.

The example shows the setup function doing the queue assertion

The README's worked example is short enough to read as the API reference:

var amqp = require('amqp-connection-manager');

// Create a new connection manager
var connection = amqp.connect(['amqp://localhost']);

// Ask the connection manager for a ChannelWrapper.  Specify a setup function to
// run every time we reconnect to the broker.
var channelWrapper = connection.createChannel({
  json: true,
  setup: function (channel) {
    // `channel` here is a regular amqplib `ConfirmChannel`.
    // Note that `this` here is the channelWrapper instance.
    return channel.assertQueue('rxQueueName', { durable: true });
  },
});

Two details matter. The `json: true` option tells the wrapper that everything published is a plain object and should be encoded for you, and the comment notes that the `channel` passed into the setup function is a regular amqplib `ConfirmChannel`, so everything you know about amqplib channels still applies. The wrapper is a proxy, not a replacement.

Publishing returns a Promise that settles when the message actually goes out

The same example continues into the send path, and the comment is the important part:

// Send some messages to the queue.  If we're not currently connected, these will be queued up in memory
// until we connect.  Note that `sendToQueue()` and `publish()` return a Promise which is fulfilled or rejected
// when the message is actually sent (or not sent.)
channelWrapper
  .sendToQueue('rxQueueName', { hello: 'world' })
  .then(function () {
    return console.log('Message was sent!  Hooray!');
  })
  .catch(function (err) {
    return console.log('Message was rejected...  Boo!');
  });

The promise does not resolve when you call publish, it resolves when the message is actually sent, which is the behaviour you want for a publisher that needs to know whether the broker accepted the message. And when the broker is down, the messages are queued in memory until it comes back. That is a convenience and a risk at the same time, since an in-memory queue is a loss risk if the process dies while disconnected, and it is the single most important thing to evaluate before putting this in front of real traffic.

Setup handlers can be added and removed while the channel is live

Sometimes you need to change a channel at run time, for example when you decide a channel that was listening to one kind of message should also listen to another. The README shows this as adding a new setup function to an existing ChannelWrapper:

channelWrapper.addSetup(function (channel) {
  return Promise.all([
    channel.assertQueue('my-queue', { exclusive: true, autoDelete: true }),
    channel.bindQueue('my-queue', 'my-exchange', 'create'),
    channel.consume('my-queue', handleMessage),
  ]);
});

`addSetup()` returns a Promise which resolves when the setup function is finished, or immediately if the underlying connection is not currently connected to a broker. There is a matching `removeSetup(setup, teardown)`, which runs `teardown(channel)` if the channel is currently connected to a broker and does not run the teardown at all otherwise. Both `setup` and `teardown` must either accept a callback or return a Promise. The conditional teardown is a design choice worth noting, because it means a teardown that assumes a live channel is not safe to rely on when the broker is down.

Round robin and service discovery are handled in the connect options

The connect function creates an AmqpConnectionManager which will connect to one of the URLs provided, and if a broker is unreachable or dies, the manager will try the next available broker, round-robin. The options list shows how much control there is. `heartbeatIntervalInSeconds` is the interval to send heartbeats to the broker and defaults to 5 seconds. `reconnectTimeInSeconds` is the time to wait before trying to reconnect, and if not specified it defaults to the heartbeat interval. `connectionOptions` is passed as options to the amqplib connect method. The interesting one is `findServers(callback)`, a function returning one or more servers to connect to, which the README says is handy when you are using a service discovery mechanism such as Consul or etcd, and that it can also return a Promise, and that if it is supplied then `urls` is ignored. The manager also emits events, with `connect`, `connectFailed`, `disconnect`, `blocked`, and `unblocked`.

Where amqp-connection-manager is the wrong tool

Two limits are structural rather than fixable. The first is the in-memory buffer. The documented behaviour is that if messages are sent while the broker is unavailable, they are queued in memory until the reconnect. For a consumer that only reads, that is fine. For a publisher that must not lose an order or an audit event, it is not acceptable, and no amount of configuration in this library changes it, because the buffer is the design rather than an oversight. The second is the wrapper boundary. Since the library is very un-opinionated and a thin wrapper, everything amqplib does not cover stays your problem, including backpressure behaviour, publisher confirms as a first class contract, and channel-level error handling beyond the `error` event. A team that wants a different trade, for example a bounded queue with backpressure, or disk persistence of undelivered messages, needs a different tool.

Editorial conclusion

amqp-connection-manager fits a Node service that consumes or produces from RabbitMQ and has been bitten by a broker restart taking the process down with it, or by a client that cannot survive a cluster with more than one broker. It is the wrong tool if your workload needs delivery guarantees across a broker outage, because the documented behaviour is to queue messages in memory, which is a loss risk when the process dies, and it is the wrong tool if you want per-message publishing confirmations as a contract, since the returned Promise resolves only when the message is actually sent. Before you adopt it, read the `connect` options carefully, particularly `findServers` for service discovery and the reconnect timing options, and check the amqplib version your application already uses against the peer dependency this package expects.

Frequently asked questions

Official sources

1. Issues
2. jwalton/node-amqp-connection-manager on GitHub
3. README
4. Releases