lib/amqpjs.js
function setupConn(connectCfg, options) {
    var conn = amqp.createConnection(connectCfg, options);
    conn.on('ready', function() {
      debug('received ready event from node-amqp');
      var eventName = 'connection';
      //Ready event will be emitted when re-connected.
      //To keep backward compatible, only emit 'connection' event for the first time
      if (_connection) {
        eventName = 'reconnect';
      }
      _connection = conn;
      _connection._isClosed = false;
      self.emit('ready');
      // wrapped in 'nextTick' for unit test friendliness
      process.nextTick(function() {
        debug('going to emit ' + eventName);
        self.emit(eventName);
      });

      autoSubscribe();
      publishCachedMessages();
      _rpcReplyQ = null;
      _rpcReplyQName = null;
      _rpcSubscribed = false;
      if (_.size(_rpcCallbacks) > 0) {
        debug('found ' + _.size(_rpcCallbacks) + ' callbacks left after reconnect');
        //still have callbacks that are waiting for response.However, since we have reconnected, they won't receive the messages
        _.each(_rpcCallbacks, function(cb) {
          return cb('connection_error');
        });
      }
      _rpcCallbacks = {};
    });
    conn.on('error', function(err) {
      self.emit('error', err);
    });
    conn.on('close', function() {
      if (_connection) {
        _connection._isClosed = true;
      }
    });
  }
function setupConn(connectCfg, options) {
    var conn = amqp.createConnection(connectCfg, options);
    conn.on('ready', function() {
      debug('received ready event from node-amqp');
      var eventName = 'connection';
      //Ready event will be emitted when re-connected.
      //To keep backward compatible, only emit 'connection' event for the first time
      if (_connection) {
        eventName = 'reconnect';
      }
      _connection = conn;
      _connection._isClosed = false;
      self.emit('ready');
      // wrapped in 'nextTick' for unit test friendliness
      process.nextTick(function() {
        debug('going to emit ' + eventName);
        self.emit(eventName);
      });

      autoSubscribe();
      publishCachedMessages();
      _rpcReplyQ = null;
      _rpcReplyQName = null;
      _rpcSubscribed = false;
      if (_.size(_rpcCallbacks) > 0) {
        debug('found ' + _.size(_rpcCallbacks) + ' callbacks left after reconnect');
        //still have callbacks that are waiting for response.However, since we have reconnected, they won't receive the messages
        _.each(_rpcCallbacks, function(cb) {
          return cb('connection_error');
        });
      }
      _rpcCallbacks = {};
    });
    conn.on('error', function(err) {
      self.emit('error', err);
    });
    conn.on('close', function() {
      if (_connection) {
        _connection._isClosed = true;
      }
    });
  }
function setupConn ( connectCfg , options ) { var conn = amqp . createConnection ( connectCfg , options ) ; conn . on ( 'ready' , function ( ) { debug ( 'received ready event from node-amqp' ) ; var eventName = 'connection' ; //Ready event will be emitted when re-connected. //To keep backward compatible, only emit 'connection' event for the first time if ( _connection ) { eventName = 'reconnect' ; } _connection = conn ; _connection . _isClosed = false ; self . emit ( 'ready' ) ; // wrapped in 'nextTick' for unit test friendliness process . nextTick ( function ( ) { debug ( 'going to emit ' + eventName ) ; self . emit ( eventName ) ; } ) ; autoSubscribe ( ) ; publishCachedMessages ( ) ; _rpcReplyQ = null ; _rpcReplyQName = null ; _rpcSubscribed = false ; if ( _ . size ( _rpcCallbacks ) > 0 ) { debug ( 'found ' + _ . size ( _rpcCallbacks ) + ' callbacks left after reconnect' ) ; //still have callbacks that are waiting for response.However, since we have reconnected, they won't receive the messages _ . each ( _rpcCallbacks , function ( cb ) { return cb ( 'connection_error' ) ; } ) ; } _rpcCallbacks = { } ; } ) ; conn . on ( 'error' , function ( err ) { self . emit ( 'error' , err ) ; } ) ; conn . on ( 'close' , function ( ) { if ( _connection ) { _connection . _isClosed = true ; } } ) ; }
Private functions 
Setup the connection
@param  {Object} connectCfg AMQP connection configurations.
@param  {Object} options    AMQP connection options.
See https://github.com/postwait/node-amqp#connection-options-and-url for the format of the connectCfg and options
Private functions Setup the connection