In all of those cases, however, the application could still communicate with LavinMQ. But what happens if the connection itself fails? For example, what happens if the network connection between the worker and LavinMQ is interrupted?
In this part, we’ll compare how two Node.js AMQP clients, amqplib and amqp-client.js, approach connection recovery. This is also where the broker abstraction we’ve used throughout the series becomes particularly useful.
Our worker communicates with LavinMQ through broker.js. The worker can focus on processing transactions, while the broker layer handles client-specific concerns such as connections, subscriptions, acknowledgements, and connection recovery. We’ll now look inside that broker layer and compare how the two clients implement it.
Connections and channels
For our Node.js application to communicate with LavinMQ, it first needs an AMQP connection. The connection provides the communication path between the application and LavinMQ.
AMQP channels operate over that connection. Instead of opening a separate TCP connection for every messaging operation, applications can use lightweight channels for operations such as publishing messages, consuming messages, and sending acknowledgements.
This becomes particularly important when discussing connection failures. If the underlying connection is lost, the application needs a way to restore its messaging operations so the worker can continue consuming transactions. How much of that recovery the application needs to implement depends on the client we use.
The two Node.js clients
amqplib is an open-source Node.js AMQP 0-9-1 client. It exposes concepts such as connections and channels directly to the application.
amqp-client.js is also an open-source Node.js AMQP 0-9-1 client. It provides both low-level and high-level APIs.
The low-level API gives developers more direct control over connections and channels. The high-level API, AMQPSession, manages more of the connection lifecycle for the application, including connection recovery. For the rest of this article, we’ll compare amqplib with the high-level AMQPSession API.
The amqplib broker implementation
Let’s first look at how our amqplib broker layer establishes its connection to LavinMQ. Inside broker.js, we explicitly create a connection and then create a channel.
connection = await amqp.connect(config.amqpUrl);
channel = await connection.createChannel();
The broker layer then uses that channel for messaging operations. For example, when the worker starts, it calls:
await broker.connect();
await broker.subscribe(handleMessage);
The worker doesn’t need to create an amqplib connection or channel itself. Those details are encapsulated in broker.js.
Conceptually, the flow looks like this:
worker.js
│
│ broker.connect()
│ broker.subscribe(...)
▼
amqplib/broker.js
│
├── connection
├── channel
└── consumer
│
▼
LavinMQ
This works while the connection is healthy. But what happens when it disappears?
Handling connection failure with amqplib
amqplib doesn’t automatically recreate the application’s connection and consumer when the connection is lost.
See it in action
Start the worker:
npm run worker:amqplib
Start the producer:
npm run producer:amqplib -- valid
The worker receives and processes the transaction normally. Now interrupt the network connection. The worker remains disconnected.
The important point is that automatic recovery isn’t provided by amqplib itself. If we want our application to recover automatically, we need to implement that behavior. In our broker layer, we can listen for the connection to close.
connection.on("close", reconnect);
connection.on("error", (error) => {
log("Connection error", {
error: error.message,
});
});
When the connection closes unexpectedly, the broker can schedule a reconnect.
function reconnect() {
if (shuttingDown || !messageHandler) return;
log(
"Connection closed. Reconnecting in 5 seconds..."
);
setTimeout(async () => {
try {
await connect();
await subscribe(messageHandler);
log("Reconnected to LavinMQ");
} catch (error) {
log("Reconnect failed", {
error: error.message,
});
reconnect();
}
}, 5000);
}
There is an important detail here. Recovering isn’t only about opening another TCP connection. The old channel belonged to the old connection. Once that connection is gone, we need to recreate the messaging state required by the application.
In our case, the broker layer:
- establishes a new connection,
- creates a new channel,
- configures the consumer again, and
- subscribes the worker’s message handler again.
The recovery flow therefore looks roughly like this:
The amqp-client.js broker implementation
Now let’s look at the same responsibility using the high-level API of amqp-client.js. Instead of explicitly creating a connection and channel, the broker establishes an AMQPSession:
session = await AMQPSession.connect(
config.amqpUrl,
{
reconnectInterval: 1000,
maxReconnectInterval: 30000,
backoffMultiplier: 2,
maxRetries: 0,
onconnect: () =>
log("amqp-client connected"),
ondisconnect: (error) =>
log("amqp-client disconnected", {
error: error?.message,
}),
onfailed: (error) =>
log("amqp-client failed to reconnect", {
error: error?.message,
}),
}
);
The broker layer then accesses the transaction queue through the session and subscribes the worker’s handler. Conceptually:
worker.js
│
│ broker.connect()
│ broker.subscribe(...)
▼
amqp-client/broker.js
│
▼
AMQPSession
│
▼
LavinMQ
The application still has a broker layer, but the implementation behind it is different.
Handling connection failure with AMQPSession
AMQPSession provides connection recovery as part of the high-level client API. If the connection is interrupted, the session can attempt to reconnect according to its configured reconnect settings.
In our example:
reconnectInterval: 1000,
maxReconnectInterval: 30000,
backoffMultiplier: 2,
maxRetries: 0,
The reconnect delay starts at one second and can increase between unsuccessful attempts, up to the configured maximum interval. The session also manages recovery of its subscriptions, so the application doesn’t need to implement the same reconnect-and-resubscribe loop we wrote for amqplib.
The recovery flow becomes:
The important difference isn’t that one client can recover and the other cannot. Both can be used to build an application that recovers from connection failures. The difference is where the recovery responsibility lives.
With amqplib, we implement the recovery behavior in our broker layer.
With high-level AMQPSession, much of that connection and subscription recovery is handled by the client.
See it in action
Start the producer:
npm run producer:amqplib -- valid
Start the worker:
npm run worker:amqplib
The worker receives and processes the transaction normally. Now interrupt the network connection. The session detects that the connection has been lost and starts its reconnect process. Restore the network connection.
Once connectivity returns, AMQPSession reconnects and restores the subscription, allowing the worker to continue receiving transactions without being restarted manually. You can publish another transaction after restoring the connection:
npm run producer:amqp-client -- valid
The worker should receive and process it after the connection has recovered.
Comparing the two approaches
The same banking application can use either client because the client-specific behavior is kept behind our broker abstraction. What changes is the implementation of that layer:
amqplib |
High-level amqp-client.js |
|
|---|---|---|
| Establish connection | amqp.connect() |
AMQPSession.connect() |
| Channel management | Application manages the channel | Managed through the session |
| Connection recovery | Application implements it | Built into AMQPSession |
| Reconnect backoff | Application implements it | Built in and configurable |
| Consumer recovery | Application restores the consumer | Subscription recovery is handled by the session |
| Low-level control | Yes | Also available through the low-level API |
With amqplib, our broker layer contains more lifecycle management:
amqplib/broker.js
connect
↓
create channel
↓
subscribe
↓
connection lost
↓
detect failure
↓
wait
↓
reconnect
↓
create channel again
↓
restore subscription
With the high-level AMQPSession API:
amqp-client/broker.js
create session
↓
subscribe
↓
connection lost
↓
AMQPSession reconnects
↓
subscription recovered
Neither approach changes what our banking worker is trying to accomplish. The worker still processes transactions, decides whether failures are retryable, protects against duplicate processing, and acknowledges completed messages through the broker layer. What changes is how the broker layer manages communication with LavinMQ.
Why the broker abstraction matters
This comparison also shows why we separated client-specific messaging operations from the application logic in Part 1. Our worker interacts with a small broker interface:
await broker.connect();
await broker.subscribe(handleMessage);
During processing, it uses operations such as:
await broker.retry(
transaction,
retries + 1
);
await broker.sendToDeadQueue(
transaction,
error,
retries
);
broker.acknowledge(message);
The banking logic doesn’t need to know whether the underlying implementation uses an explicit amqplib connection and channel or an AMQPSession.
This doesn’t make the two clients identical. Their APIs, message representations, connection lifecycle, and recovery behavior differ. What it does is keep those differences concentrated in the broker-specific part of the application instead of spreading them throughout the banking logic.
Summary
In this part, we looked at a different type of failure from the transaction failures we handled in Part 2 — losing the connection between our application and LavinMQ.
With amqplib, connections and channels are exposed directly. If we want automatic recovery, our application needs to detect the lost connection, establish a new one, recreate the channel, and restore the consumer.
With the high-level AMQPSession API from amqp-client.js, more of that lifecycle is handled by the client. The session provides configurable reconnection and restores its subscriptions when the connection recovers.
The important difference is not whether recovery is possible with either client. It is who is responsible for implementing it.
Because we’ve isolated these client-specific responsibilities behind broker.js, changing clients doesn’t require us to redesign the banking application’s reliability logic.
In Part 4, we’ll put that architecture to use. We’ll migrate the broker layer from amqplib to amqp-client.js and look at what actually changes — and what can stay the same.
Hailay Gebremeskel