Reconnect with RabbitMQ using amqplib after rabbitMQ restarts

Question

I have this rabbitmq connection code using which I have connected to the queue and using it.

const amqp = require('amqplib');

class RabbitMq {
    constructor(connStr) {
      this.connStr = connStr;
      this.conn = null;
      this.channel = null;
      this.queues = {};
      this.isConnecting = false;
  }
async connect() {
    if (this.conn) return;
    this.conn = await amqp.connect(this.connStr);
  }

async getChannel() {
    if (this.channel) return this.channel;
    if (!this.conn) await this.connect();
    this.channel = await this.conn.createChannel();
    return this.channel;
  }

async sendToQueue(queue, message) {
    if (!this.channel) await this.getChannel();
    return this.channel.sendToQueue(
        queue,
        Buffer.from(JSON.stringify(message))
    );
  }

async consume(queue, handler) {
    if (!this.conn) await this.connect();
    if (!this.channel) await this.getChannel();
    await this.channel.assertQueue(queue);
    this.channel.prefetch(1);
    this.channel.consume(queue, handler);
  }

async assertQueue(queue) {
    if (!this.channel) await this.getChannel();
    await this.channel.assertQueue(queue);
  }

async ack(message) {
    if (!this.channel) await this.getChannel();
    this.channel.ack(message);
  }
}

// handle SIGTERM, SIGINT
module.exports = new RabbitMq(process.env.CLOUDAMQP_URL);

Now this works fine but as soon as I run this command on terminal,

brew services restart rabbitmq

which is basically restarting the rabbitMQ, I get this heartbeat error and then I m just unable to connect to rabbitMQ again. This keeps on happening. I also tried to put try catch in the connect() function but its not getting caught there. The error I get is:

Error: Heartbeat timeout
at Heart.<anonymous> (/Users/..../node_modules/amqplib/lib/connection.js:427:19)
at Heart.emit (events.js:315:20)
at Heart.runHeartbeat (/Users/..../node_modules/amqplib/lib/heartbeat.js:88:17)
at listOnTimeout (internal/timers.js:549:17)
at processTimers (internal/timers.js:492:7)

Any thoughts? If there is a way I can catch this exception then may be I can call connect again from there.

Reconnect with RabbitMQ using amqplib after rabbitMQ restarts

0 Answers0