How to use the amqplib/callback_api.connect function in amqplib

let spammer = function(obj) {
  let {type, exchange, binding, message, uri} = obj;

  amqp.connect(uri, (err, conn) => {
    conn.createChannel((err, ch) => {
      
      ch.assertExchange(exchange, type, {durable: false});
      ch.publish(exchange, binding, new Buffer.from(message));
      console.log('.-. message sent', message)
      
    });
    // setTimeout(function() { conn.close(); process.exit(0) }, 500)
    setTimeout(function() { conn.close() }, 1000)
  }) 
}

function on_channel_open(err, ch) {
    ch.assertQueue(q, {durable: false}, function(err, ok) {
      if (err !== null) return bail(err, conn);
      ch.consume(q, function(msg) { // message callback
        console.log(" [x] Received '%s'", msg.content.toString());
      }, {noAck: true}, function(_consumeOk) { // consume callback
        console.log(' [*] Waiting for messages. To exit press CTRL+C');
      });
    });
  }

  conn.createChannel(on_channel_open);
}

amqp.connect(on_connect);

self._connection_callbacks.push(callback)

    if (!first) {
        return
    }

    let connection_options = {}

    if (self._service_name) {
        connection_options.clientProperties = {
            connection_name: self._service_name
        }
    }

    // So let's connect!
    amqplib.connect(self._url, connection_options, (err, connection) => {
        if (err) {
            throw err
        }

        // Everything's go fine, so we'll set this global
        // object to our new connection.
        self._connection = connection

        // Time to run the callbacks. Let's run them and
        // take them out of the queue.
        // Loop through and make everything happen!
        while (self._connection_callbacks.length > 0) {
            self._connection_callbacks[0]()
            self._connection_callbacks.shift()
        }
    })

}

function on_connect(err, conn) {
  if (err !== null) return bail(err);
  var ex = 'topic_logs', exopts = {durable: false};
  conn.createChannel(function(err, ch) {
    ch.assertExchange(ex, 'topic', exopts, function(err, ok) {
      if (err !== null) return bail(err, conn);
      ch.publish(ex, key, Buffer.from(message));
      console.log(" [x] Sent %s:'%s'", key, message);
      ch.close(function() { conn.close(); });
    });
  });
}

amqp.connect(on_connect);

ch.consume(queue, logMessage, {noAck: true}, function(err) {
        if (err !== null) return bail(err, conn);
        console.log(' [*] Waiting for logs. To exit press CTRL+C.');
        sub(null);
      });
    });
  });
}

function logMessage(msg) {
  console.log(" [x] %s:'%s'",
              msg.fields.routingKey,
              msg.content.toString());
}

amqp.connect(on_connect);

var msg = 'Hello World!';

  function on_channel_open(err, ch) {
    if (err !== null) return bail(err, conn);
    ch.assertQueue(q, {durable: false}, function(err, ok) {
      if (err !== null) return bail(err, conn);
      ch.sendToQueue(q, new Buffer(msg));
      console.log(" [x] Sent '%s'", msg);
      ch.close(function() { conn.close(); });
    });
  }

  conn.createChannel(on_channel_open);
}

amqp.connect(on_connect);

#!/usr/bin/env node

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function(error1, channel) {
        if (error1) {
            throw error1;
        }
        var queue = 'rpc_queue';

        channel.assertQueue(queue, {
            durable: false
        });
        channel.prefetch(1);
        console.log(' [x] Awaiting RPC requests');
        channel.consume(queue, function reply(msg) {
            var n = parseInt(msg.content.toString());

function triggerWebhook(ballot, status, callback) {
  require('amqplib/callback_api').connect('amqp://localhost', (err, conn) => {
    if (err) {
      log.error('Unable to connect to AMQP service.');
      return callback(err, null);
    }
    return conn.createChannel((err2, ch) => {
      if (err2) {
        log.error('Unable to create new AMQP channel.');
        return callback(err2, null);
      }
      msg = {
        url: ballot.app.webhookURL,
        event: {
          app: ballot.app,
          vote: {id: ballot.vote.id},
          user: {sub: ballot.user.sub},
          status: status,

var connectAmqp = () => {
  amqp.connect(rabbitUrl, function (err, conn) {
    if (err) {
      log.debug(log.defaultContext(), 'Error in connecting to rabbit url: ' + rabbitUrl);
      log.debug(log.defaultContext(), err);
      return setTimeout(connectAmqp, 1000);
    }
    initChannel(conn);
  });
};