src/PubConnection.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)

    let _messagesToResend = []
    let _onMessagePublished = (...args) => {
      if (_socket) _messagesToResend.push(args)
    }
    _publishStream.on('message', _onMessagePublished)

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

    let connectionStart = Date.now()
    const onSocketConnected = timingoutCallback((err) => {
      _publishStream.removeListener('message', _onMessagePublished)
      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`)

      _socket = newSocket

      let totalMessagesToResend = _messagesToResend.length
      if (totalMessagesToResend) {
        debug(`resending ${totalMessagesToResend} messages published while transitioning`)
        _messagesToResend.forEach(message => _socket.send(message))
      }

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

    newSocket.once('connect', () => onSocketConnected())
    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)

    let _messagesToResend = []
    let _onMessagePublished = (...args) => {
      if (_socket) _messagesToResend.push(args)
    }
    _publishStream.on('message', _onMessagePublished)

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

    let connectionStart = Date.now()
    const onSocketConnected = timingoutCallback((err) => {
      _publishStream.removeListener('message', _onMessagePublished)
      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`)

      _socket = newSocket

      let totalMessagesToResend = _messagesToResend.length
      if (totalMessagesToResend) {
        debug(`resending ${totalMessagesToResend} messages published while transitioning`)
        _messagesToResend.forEach(message => _socket.send(message))
      }

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

    newSocket.once('connect', () => onSocketConnected())
    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 ) let _messagesToResend = [ ] let _onMessagePublished = ( ... args ) => { if ( _socket ) _messagesToResend . push ( args ) } _publishStream . on ( 'message' , _onMessagePublished ) debug ( ` ${ master . name } ${ master . endpoint } ` ) let connectionStart = Date . now ( ) const onSocketConnected = timingoutCallback ( ( err ) => { _publishStream . removeListener ( 'message' , _onMessagePublished ) 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 } ` ) _socket = newSocket let totalMessagesToResend = _messagesToResend . length if ( totalMessagesToResend ) { debug ( ` ${ totalMessagesToResend } ` ) _messagesToResend . forEach ( message => _socket . send ( message ) ) } if ( previousSocket ) { previousSocket . close ( ) debug ( ` ${ previousSocket . _master . name } ${ previousSocket . _master . endpoint } ` ) } else { connection . emit ( 'connect' ) } } , 500 ) newSocket . once ( 'connect' , ( ) => onSocketConnected ( ) ) return connection }
creates a new pub socket and tries to connect it to the provided master;
after connection the socket is used to publish messages in the bus
and an eventual previous socket is closed
@param  {object} master
@return {object} the publishConnection instance
creates a new pub socket and tries to connect it to the provided master ; after connection the socket is used to publish messages in the bus and an eventual previous socket is closed