src/SubConnection.js
function connect (master) {
    if (_socket && _socket._master.name === master.name) {
      debug(`already connected to master ${master.name}`)
      return
    }
    if (_connectingMaster && _connectingMaster.name === master.name) {
      debug(`already connecting to master ${master.name}`)
      return
    }

    _connectingMaster = master
    let newSocket = _getSocket(master)

    debug(`connecting to ${master.name} at ${master.endpoint}`)

    let connectionStart = Date.now()
    const onSocketReceiving = timingoutCallback((err, ...args) => {
      newSocket.unmonitor()

      if (
        err ||
        !_connectingMaster ||
        _connectingMaster.name === !master.name
      ) {
        newSocket.close()
        if (err) {
          debug(`failed to connect to ${master.name} at ${master.endpoint}`)
          disconnect()
        }
        return
      }

      _connectingMaster = null
      let previousSocket = _socket

      debug(`${previousSocket ? 'switched' : 'connected'} to ${master.name} at ${master.endpoint} in ${Date.now() - connectionStart} ms`)

      if (args.length) {
        _onSocketMessage(...args)
      }

      _socket = newSocket
      _socket.removeAllListeners()
      _subscribedChannels.forEach(channel => _socket.subscribe(channel))
      _socket.on('message', _onSocketMessage)
      _lastHeartbeatReceivedTime = Date.now()
      _monitorHeartbeats()

      if (previousSocket) {
        setTimeout(() => {
          previousSocket.removeAllListeners()
          previousSocket.close()
          debug(`closed previous connection to ${previousSocket._master.name} at ${previousSocket._master.endpoint}`)
        }, 300)
      } else {
        connection.emit('connect')
      }
    }, 500)

    newSocket.once('connect', () => onSocketReceiving())
    newSocket.once('message', (...args) => onSocketReceiving(null, ...args))
    return connection
  }
function connect (master) {
    if (_socket && _socket._master.name === master.name) {
      debug(`already connected to master ${master.name}`)
      return
    }
    if (_connectingMaster && _connectingMaster.name === master.name) {
      debug(`already connecting to master ${master.name}`)
      return
    }

    _connectingMaster = master
    let newSocket = _getSocket(master)

    debug(`connecting to ${master.name} at ${master.endpoint}`)

    let connectionStart = Date.now()
    const onSocketReceiving = timingoutCallback((err, ...args) => {
      newSocket.unmonitor()

      if (
        err ||
        !_connectingMaster ||
        _connectingMaster.name === !master.name
      ) {
        newSocket.close()
        if (err) {
          debug(`failed to connect to ${master.name} at ${master.endpoint}`)
          disconnect()
        }
        return
      }

      _connectingMaster = null
      let previousSocket = _socket

      debug(`${previousSocket ? 'switched' : 'connected'} to ${master.name} at ${master.endpoint} in ${Date.now() - connectionStart} ms`)

      if (args.length) {
        _onSocketMessage(...args)
      }

      _socket = newSocket
      _socket.removeAllListeners()
      _subscribedChannels.forEach(channel => _socket.subscribe(channel))
      _socket.on('message', _onSocketMessage)
      _lastHeartbeatReceivedTime = Date.now()
      _monitorHeartbeats()

      if (previousSocket) {
        setTimeout(() => {
          previousSocket.removeAllListeners()
          previousSocket.close()
          debug(`closed previous connection to ${previousSocket._master.name} at ${previousSocket._master.endpoint}`)
        }, 300)
      } else {
        connection.emit('connect')
      }
    }, 500)

    newSocket.once('connect', () => onSocketReceiving())
    newSocket.once('message', (...args) => onSocketReceiving(null, ...args))
    return connection
  }
function connect ( master ) { if ( _socket && _socket . _master . name === master . name ) { debug ( ` ${ master . name } ` ) return } if ( _connectingMaster && _connectingMaster . name === master . name ) { debug ( ` ${ master . name } ` ) return } _connectingMaster = master let newSocket = _getSocket ( master ) debug ( ` ${ master . name } ${ master . endpoint } ` ) let connectionStart = Date . now ( ) const onSocketReceiving = timingoutCallback ( ( err , ... args ) => { newSocket . unmonitor ( ) if ( err || ! _connectingMaster || _connectingMaster . name === ! master . name ) { newSocket . close ( ) if ( err ) { debug ( ` ${ master . name } ${ master . endpoint } ` ) disconnect ( ) } return } _connectingMaster = null let previousSocket = _socket debug ( ` ${ previousSocket ? 'switched' : 'connected' } ${ master . name } ${ master . endpoint } ${ Date . now ( ) - connectionStart } ` ) if ( args . length ) { _onSocketMessage ( ... args ) } _socket = newSocket _socket . removeAllListeners ( ) _subscribedChannels . forEach ( channel => _socket . subscribe ( channel ) ) _socket . on ( 'message' , _onSocketMessage ) _lastHeartbeatReceivedTime = Date . now ( ) _monitorHeartbeats ( ) if ( previousSocket ) { setTimeout ( ( ) => { previousSocket . removeAllListeners ( ) previousSocket . close ( ) debug ( ` ${ previousSocket . _master . name } ${ previousSocket . _master . endpoint } ` ) } , 300 ) } else { connection . emit ( 'connect' ) } } , 500 ) newSocket . once ( 'connect' , ( ) => onSocketReceiving ( ) ) newSocket . once ( 'message' , ( ... args ) => onSocketReceiving ( null , ... args ) ) return connection }
creates a new sub socket and tries to connect it to the provided master;
after connection the socket is used to receive messages from the bus
and an eventual previous socket is closed
@param  {object} master
@return {object} the subscribeConnection instance
creates a new sub socket and tries to connect it to the provided master ; after connection the socket is used to receive messages from the bus and an eventual previous socket is closed