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,30 +14,11 @@
 
     intervalID = clearInterval(intervalID);
 
-    return createChannel();
-  });
-}
-
-function createChannel() {
-  if (amqpConnection) {
-    return amqpConnection.createChannel()
+    return _connection.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
