lib/topologies/replset.js
function(host, self, options) {
  // If this is not the initial scan
  // Is this server already being monitoried, then skip monitoring
  if (!options.haInterval) {
    for (var i = 0; i < self.intervalIds.length; i++) {
      if (self.intervalIds[i].__host === host) {
        return;
      }
    }
  }

  // Get the haInterval
  var _process = options.haInterval ? Timeout : Interval;
  var _haInterval = options.haInterval ? options.haInterval : self.s.haInterval;

  // Create the interval
  var intervalId = new _process(function() {
    if (self.state === DESTROYED || self.state === UNREFERENCED) {
      // clearInterval(intervalId);
      intervalId.stop();
      return;
    }

    // Do we already have server connection available for this host
    var _server = self.s.replicaSetState.get(host);

    // Check if we have a known server connection and reuse
    if (_server) {
      // Ping the server
      return pingServer(self, _server, function(err) {
        if (err) {
          // NOTE: should something happen here?
          return;
        }

        if (self.state === DESTROYED || self.state === UNREFERENCED) {
          intervalId.stop();
          return;
        }

        // Filter out all called intervaliIds
        self.intervalIds = self.intervalIds.filter(function(intervalId) {
          return intervalId.isRunning();
        });

        // Initial sweep
        if (_process === Timeout) {
          if (
            self.state === CONNECTING &&
            ((self.s.replicaSetState.hasSecondary() &&
              self.s.options.secondaryOnlyConnectionAllowed) ||
              self.s.replicaSetState.hasPrimary())
          ) {
            self.state = CONNECTED;

            // Emit connected sign
            process.nextTick(function() {
              self.emit('connect', self);
            });

            // Start topology interval check
            topologyMonitor(self, {});
          }
        } else {
          if (
            self.state === DISCONNECTED &&
            ((self.s.replicaSetState.hasSecondary() &&
              self.s.options.secondaryOnlyConnectionAllowed) ||
              self.s.replicaSetState.hasPrimary())
          ) {
            self.state = CONNECTED;

            // Rexecute any stalled operation
            rexecuteOperations(self);

            // Emit connected sign
            process.nextTick(function() {
              self.emit('reconnect', self);
            });
          }
        }

        if (
          self.initialConnectState.connect &&
          !self.initialConnectState.fullsetup &&
          self.s.replicaSetState.hasPrimaryAndSecondary()
        ) {
          // Set initial connect state
          self.initialConnectState.fullsetup = true;
          self.initialConnectState.all = true;

          process.nextTick(function() {
            self.emit('fullsetup', self);
            self.emit('all', self);
          });
        }
      });
    }
  }, _haInterval);

  // Start the interval
  intervalId.start();
  // Add the intervalId host name
  intervalId.__host = host;
  // Add the intervalId to our list of intervalIds
  self.intervalIds.push(intervalId);
}
function(host, self, options) {
  // If this is not the initial scan
  // Is this server already being monitoried, then skip monitoring
  if (!options.haInterval) {
    for (var i = 0; i < self.intervalIds.length; i++) {
      if (self.intervalIds[i].__host === host) {
        return;
      }
    }
  }

  // Get the haInterval
  var _process = options.haInterval ? Timeout : Interval;
  var _haInterval = options.haInterval ? options.haInterval : self.s.haInterval;

  // Create the interval
  var intervalId = new _process(function() {
    if (self.state === DESTROYED || self.state === UNREFERENCED) {
      // clearInterval(intervalId);
      intervalId.stop();
      return;
    }

    // Do we already have server connection available for this host
    var _server = self.s.replicaSetState.get(host);

    // Check if we have a known server connection and reuse
    if (_server) {
      // Ping the server
      return pingServer(self, _server, function(err) {
        if (err) {
          // NOTE: should something happen here?
          return;
        }

        if (self.state === DESTROYED || self.state === UNREFERENCED) {
          intervalId.stop();
          return;
        }

        // Filter out all called intervaliIds
        self.intervalIds = self.intervalIds.filter(function(intervalId) {
          return intervalId.isRunning();
        });

        // Initial sweep
        if (_process === Timeout) {
          if (
            self.state === CONNECTING &&
            ((self.s.replicaSetState.hasSecondary() &&
              self.s.options.secondaryOnlyConnectionAllowed) ||
              self.s.replicaSetState.hasPrimary())
          ) {
            self.state = CONNECTED;

            // Emit connected sign
            process.nextTick(function() {
              self.emit('connect', self);
            });

            // Start topology interval check
            topologyMonitor(self, {});
          }
        } else {
          if (
            self.state === DISCONNECTED &&
            ((self.s.replicaSetState.hasSecondary() &&
              self.s.options.secondaryOnlyConnectionAllowed) ||
              self.s.replicaSetState.hasPrimary())
          ) {
            self.state = CONNECTED;

            // Rexecute any stalled operation
            rexecuteOperations(self);

            // Emit connected sign
            process.nextTick(function() {
              self.emit('reconnect', self);
            });
          }
        }

        if (
          self.initialConnectState.connect &&
          !self.initialConnectState.fullsetup &&
          self.s.replicaSetState.hasPrimaryAndSecondary()
        ) {
          // Set initial connect state
          self.initialConnectState.fullsetup = true;
          self.initialConnectState.all = true;

          process.nextTick(function() {
            self.emit('fullsetup', self);
            self.emit('all', self);
          });
        }
      });
    }
  }, _haInterval);

  // Start the interval
  intervalId.start();
  // Add the intervalId host name
  intervalId.__host = host;
  // Add the intervalId to our list of intervalIds
  self.intervalIds.push(intervalId);
}
function ( host , self , options ) { // If this is not the initial scan // Is this server already being monitoried, then skip monitoring if ( ! options . haInterval ) { for ( var i = 0 ; i < self . intervalIds . length ; i ++ ) { if ( self . intervalIds [ i ] . __host === host ) { return ; } } } // Get the haInterval var _process = options . haInterval ? Timeout : Interval ; var _haInterval = options . haInterval ? options . haInterval : self . s . haInterval ; // Create the interval var intervalId = new _process ( function ( ) { if ( self . state === DESTROYED || self . state === UNREFERENCED ) { // clearInterval(intervalId); intervalId . stop ( ) ; return ; } // Do we already have server connection available for this host var _server = self . s . replicaSetState . get ( host ) ; // Check if we have a known server connection and reuse if ( _server ) { // Ping the server return pingServer ( self , _server , function ( err ) { if ( err ) { // NOTE: should something happen here? return ; } if ( self . state === DESTROYED || self . state === UNREFERENCED ) { intervalId . stop ( ) ; return ; } // Filter out all called intervaliIds self . intervalIds = self . intervalIds . filter ( function ( intervalId ) { return intervalId . isRunning ( ) ; } ) ; // Initial sweep if ( _process === Timeout ) { if ( self . state === CONNECTING && ( ( self . s . replicaSetState . hasSecondary ( ) && self . s . options . secondaryOnlyConnectionAllowed ) || self . s . replicaSetState . hasPrimary ( ) ) ) { self . state = CONNECTED ; // Emit connected sign process . nextTick ( function ( ) { self . emit ( 'connect' , self ) ; } ) ; // Start topology interval check topologyMonitor ( self , { } ) ; } } else { if ( self . state === DISCONNECTED && ( ( self . s . replicaSetState . hasSecondary ( ) && self . s . options . secondaryOnlyConnectionAllowed ) || self . s . replicaSetState . hasPrimary ( ) ) ) { self . state = CONNECTED ; // Rexecute any stalled operation rexecuteOperations ( self ) ; // Emit connected sign process . nextTick ( function ( ) { self . emit ( 'reconnect' , self ) ; } ) ; } } if ( self . initialConnectState . connect && ! self . initialConnectState . fullsetup && self . s . replicaSetState . hasPrimaryAndSecondary ( ) ) { // Set initial connect state self . initialConnectState . fullsetup = true ; self . initialConnectState . all = true ; process . nextTick ( function ( ) { self . emit ( 'fullsetup' , self ) ; self . emit ( 'all' , self ) ; } ) ; } } ) ; } } , _haInterval ) ; // Start the interval intervalId . start ( ) ; // Add the intervalId host name intervalId . __host = host ; // Add the intervalId to our list of intervalIds self . intervalIds . push ( intervalId ) ; }
Each server is monitored in parallel in their own timeout loop
Each server is monitored in parallel in their own timeout loop