var amqp = require('amqplib');

var amqpUrl, amqpConnection, intervalID;

function connect(_amqpUrl) {
  amqpUrl = amqpUrl || _amqpUrl || process.env.AMQP_URL || 'amqp://localhost';

  return amqp.connect(amqpUrl)
  .then(function (_connection) {
    amqpConnection = _connection;

    _connection.on('close', reconnect);
    _connection.on('error', reconnect);

    intervalID = clearInterval(intervalID);

    return createChannel();
  });
}

function createChannel() {
  if (amqpConnection) {
    return amqpConnection.createChannel()
    .then(function (_channel) {
      _channel.on('close', recreateChannel);
      _channel.on('error', recreateChannel);

      intervalID = clearInterval(intervalID);

      return _channel;
    });
  } else {
    reconnect();
  }
}

function recreateChannel() {
  if (!intervalID) {
    intervalID = setInterval(createChannel, 1000);
  }
}

function reconnect() {
  if (!intervalID) {
    intervalID = setInterval(connect, 1000);
  }
}

function disconnect() {
  if (amqpConnection) {
    amqpConnection.close();
  }
}

module.exports = function () {
  return {
    connect: connect,
    reconnect: reconnect,
    disconnect: disconnect
  };
};

var amqp = require('amqplib');

var amqpUrl, amqpConnection, intervalID;

function connect(_amqpUrl) {
  amqpUrl = amqpUrl || _amqpUrl || process.env.AMQP_URL || 'amqp://localhost';

  return amqp.connect(amqpUrl)
  .then(function (_connection) {
    amqpConnection = _connection;

    _connection.on('close', reconnect);
    _connection.on('error', reconnect);

    intervalID = clearInterval(intervalID);

    return _connection.createChannel()
    .then(function (_channel) {
      return _channel;
    });
  });
}

function reconnect() {
  if (!intervalID) {
    intervalID = setInterval(connect, 1000);
  }
}

function disconnect() {
  if (amqpConnection) {
    amqpConnection.close();
  }
}

module.exports = function () {
  return {
    connect: connect,
    reconnect: reconnect,
    disconnect: disconnect
  };
};

--- 
+++ 
@@ -14,11 +14,30 @@
 
     intervalID = clearInterval(intervalID);
 
-    return _connection.createChannel()
+    return createChannel();
+  });
+}
+
+function createChannel() {
+  if (amqpConnection) {
+    return amqpConnection.createChannel()
     .then(function (_channel) {
+      _channel.on('close', recreateChannel);
+      _channel.on('error', recreateChannel);
+
+      intervalID = clearInterval(intervalID);
+
       return _channel;
     });
-  });
+  } else {
+    reconnect();
+  }
+}
+
+function recreateChannel() {
+  if (!intervalID) {
+    intervalID = setInterval(createChannel, 1000);
+  }
 }
 
 function reconnect() {
Add createChannel and recreateChannel and handle channel error and close event
