First Commit

node_modules/mongodb/lib/core/sdam/monitoring.js

@@ -0,0 +1,228 @@
+'use strict';
+
+const ServerDescription = require('./server_description').ServerDescription;
+const calculateDurationInMs = require('../utils').calculateDurationInMs;
+
+/**
+ * Published when server description changes, but does NOT include changes to the RTT.
+ *
+ * @property {Object} topologyId A unique identifier for the topology
+ * @property {ServerAddress} address The address (host/port pair) of the server
+ * @property {ServerDescription} previousDescription The previous server description
+ * @property {ServerDescription} newDescription The new server description
+ */
+class ServerDescriptionChangedEvent {
+  constructor(topologyId, address, previousDescription, newDescription) {
+    Object.assign(this, { topologyId, address, previousDescription, newDescription });
+  }
+}
+
+/**
+ * Published when server is initialized.
+ *
+ * @property {Object} topologyId A unique identifier for the topology
+ * @property {ServerAddress} address The address (host/port pair) of the server
+ */
+class ServerOpeningEvent {
+  constructor(topologyId, address) {
+    Object.assign(this, { topologyId, address });
+  }
+}
+
+/**
+ * Published when server is closed.
+ *
+ * @property {ServerAddress} address The address (host/port pair) of the server
+ * @property {Object} topologyId A unique identifier for the topology
+ */
+class ServerClosedEvent {
+  constructor(topologyId, address) {
+    Object.assign(this, { topologyId, address });
+  }
+}
+
+/**
+ * Published when topology description changes.
+ *
+ * @property {Object} topologyId
+ * @property {TopologyDescription} previousDescription The old topology description
+ * @property {TopologyDescription} newDescription The new topology description
+ */
+class TopologyDescriptionChangedEvent {
+  constructor(topologyId, previousDescription, newDescription) {
+    Object.assign(this, { topologyId, previousDescription, newDescription });
+  }
+}
+
+/**
+ * Published when topology is initialized.
+ *
+ * @param {Object} topologyId A unique identifier for the topology
+ */
+class TopologyOpeningEvent {
+  constructor(topologyId) {
+    Object.assign(this, { topologyId });
+  }
+}
+
+/**
+ * Published when topology is closed.
+ *
+ * @param {Object} topologyId A unique identifier for the topology
+ */
+class TopologyClosedEvent {
+  constructor(topologyId) {
+    Object.assign(this, { topologyId });
+  }
+}
+
+/**
+ * Fired when the server monitor’s ismaster command is started - immediately before
+ * the ismaster command is serialized into raw BSON and written to the socket.
+ *
+ * @property {Object} connectionId The connection id for the command
+ */
+class ServerHeartbeatStartedEvent {
+  constructor(connectionId) {
+    Object.assign(this, { connectionId });
+  }
+}
+
+/**
+ * Fired when the server monitor’s ismaster succeeds.
+ *
+ * @param {Number} duration The execution time of the event in ms
+ * @param {Object} reply The command reply
+ * @param {Object} connectionId The connection id for the command
+ */
+class ServerHeartbeatSucceededEvent {
+  constructor(duration, reply, connectionId) {
+    Object.assign(this, { duration, reply, connectionId });
+  }
+}
+
+/**
+ * Fired when the server monitor’s ismaster fails, either with an “ok: 0” or a socket exception.
+ *
+ * @param {Number} duration The execution time of the event in ms
+ * @param {MongoError|Object} failure The command failure
+ * @param {Object} connectionId The connection id for the command
+ */
+class ServerHeartbeatFailedEvent {
+  constructor(duration, failure, connectionId) {
+    Object.assign(this, { duration, failure, connectionId });
+  }
+}
+
+/**
+ * Performs a server check as described by the SDAM spec.
+ *
+ * NOTE: This method automatically reschedules itself, so that there is always an active
+ * monitoring process
+ *
+ * @param {Server} server The server to monitor
+ */
+function monitorServer(server, options) {
+  options = options || {};
+  const heartbeatFrequencyMS = options.heartbeatFrequencyMS || 10000;
+
+  if (options.initial === true) {
+    server.s.monitorId = setTimeout(() => monitorServer(server), heartbeatFrequencyMS);
+    return;
+  }
+
+  // executes a single check of a server
+  const checkServer = callback => {
+    let start = process.hrtime();
+
+    // emit a signal indicating we have started the heartbeat
+    server.emit('serverHeartbeatStarted', new ServerHeartbeatStartedEvent(server.name));
+
+    // NOTE: legacy monitoring event
+    process.nextTick(() => server.emit('monitoring', server));
+
+    server.command(
+      'admin.$cmd',
+      { ismaster: true },
+      {
+        monitoring: true,
+        socketTimeout: server.s.options.connectionTimeout || 2000
+      },
+      (err, result) => {
+        let duration = calculateDurationInMs(start);
+
+        if (err) {
+          server.emit(
+            'serverHeartbeatFailed',
+            new ServerHeartbeatFailedEvent(duration, err, server.name)
+          );
+
+          return callback(err, null);
+        }
+
+        const isMaster = result.result;
+        server.emit(
+          'serverHeartbeatSucceded',
+          new ServerHeartbeatSucceededEvent(duration, isMaster, server.name)
+        );
+
+        return callback(null, isMaster);
+      }
+    );
+  };
+
+  const successHandler = isMaster => {
+    server.s.monitoring = false;
+
+    // emit an event indicating that our description has changed
+    server.emit('descriptionReceived', new ServerDescription(server.description.address, isMaster));
+
+    // schedule the next monitoring process
+    server.s.monitorId = setTimeout(() => monitorServer(server), heartbeatFrequencyMS);
+  };
+
+  // run the actual monitoring loop
+  server.s.monitoring = true;
+  checkServer((err, isMaster) => {
+    if (!err) {
+      successHandler(isMaster);
+      return;
+    }
+
+    // According to the SDAM specification's "Network error during server check" section, if
+    // an ismaster call fails we reset the server's pool. If a server was once connected,
+    // change its type to `Unknown` only after retrying once.
+    server.s.pool.reset(() => {
+      // otherwise re-attempt monitoring once
+      checkServer((error, isMaster) => {
+        if (error) {
+          server.s.monitoring = false;
+
+          // we revert to an `Unknown` by emitting a default description with no isMaster
+          server.emit(
+            'descriptionReceived',
+            new ServerDescription(server.description.address, null, { error })
+          );
+
+          // we do not reschedule monitoring in this case
+          return;
+        }
+
+        successHandler(isMaster);
+      });
+    });
+  });
+}
+
+module.exports = {
+  ServerDescriptionChangedEvent,
+  ServerOpeningEvent,
+  ServerClosedEvent,
+  TopologyDescriptionChangedEvent,
+  TopologyOpeningEvent,
+  TopologyClosedEvent,
+  ServerHeartbeatStartedEvent,
+  ServerHeartbeatSucceededEvent,
+  ServerHeartbeatFailedEvent,
+  monitorServer
+};

node_modules/mongodb/lib/core/sdam/server.js

@@ -0,0 +1,491 @@
+'use strict';
+const EventEmitter = require('events');
+const MongoError = require('../error').MongoError;
+const Pool = require('../connection/pool');
+const relayEvents = require('../utils').relayEvents;
+const wireProtocol = require('../wireprotocol');
+const BSON = require('../connection/utils').retrieveBSON();
+const createClientInfo = require('../topologies/shared').createClientInfo;
+const Logger = require('../connection/logger');
+const ServerDescription = require('./server_description').ServerDescription;
+const ReadPreference = require('../topologies/read_preference');
+const monitorServer = require('./monitoring').monitorServer;
+const MongoParseError = require('../error').MongoParseError;
+const MongoNetworkError = require('../error').MongoNetworkError;
+const collationNotSupported = require('../utils').collationNotSupported;
+const debugOptions = require('../connection/utils').debugOptions;
+const isSDAMUnrecoverableError = require('../error').isSDAMUnrecoverableError;
+
+// Used for filtering out fields for logging
+const DEBUG_FIELDS = [
+  'reconnect',
+  'reconnectTries',
+  'reconnectInterval',
+  'emitError',
+  'cursorFactory',
+  'host',
+  'port',
+  'size',
+  'keepAlive',
+  'keepAliveInitialDelay',
+  'noDelay',
+  'connectionTimeout',
+  'checkServerIdentity',
+  'socketTimeout',
+  'ssl',
+  'ca',
+  'crl',
+  'cert',
+  'key',
+  'rejectUnauthorized',
+  'promoteLongs',
+  'promoteValues',
+  'promoteBuffers',
+  'servername'
+];
+
+const STATE_DISCONNECTED = 0;
+const STATE_CONNECTING = 1;
+const STATE_CONNECTED = 2;
+
+/**
+ *
+ * @fires Server#serverHeartbeatStarted
+ * @fires Server#serverHeartbeatSucceeded
+ * @fires Server#serverHeartbeatFailed
+ */
+class Server extends EventEmitter {
+  /**
+   * Create a server
+   *
+   * @param {ServerDescription} description
+   * @param {Object} options
+   */
+  constructor(description, options, topology) {
+    super();
+
+    this.s = {
+      // the server description
+      description,
+      // a saved copy of the incoming options
+      options,
+      // the server logger
+      logger: Logger('Server', options),
+      // the bson parser
+      bson: options.bson || new BSON(),
+      // client metadata for the initial handshake
+      clientInfo: createClientInfo(options),
+      // state variable to determine if there is an active server check in progress
+      monitoring: false,
+      // the implementation of the monitoring method
+      monitorFunction: options.monitorFunction || monitorServer,
+      // the connection pool
+      pool: null,
+      // the server state
+      state: STATE_DISCONNECTED,
+      credentials: options.credentials,
+      topology
+    };
+  }
+
+  get description() {
+    return this.s.description;
+  }
+
+  get name() {
+    return this.s.description.address;
+  }
+
+  get autoEncrypter() {
+    if (this.s.options && this.s.options.autoEncrypter) {
+      return this.s.options.autoEncrypter;
+    }
+    return null;
+  }
+
+  /**
+   * Initiate server connect
+   */
+  connect(options) {
+    options = options || {};
+
+    // do not allow connect to be called on anything that's not disconnected
+    if (this.s.pool && !this.s.pool.isDisconnected() && !this.s.pool.isDestroyed()) {
+      throw new MongoError(`Server instance in invalid state ${this.s.pool.state}`);
+    }
+
+    // create a pool
+    const addressParts = this.description.address.split(':');
+    const poolOptions = Object.assign(
+      { host: addressParts[0], port: parseInt(addressParts[1], 10) },
+      this.s.options,
+      options,
+      { bson: this.s.bson }
+    );
+
+    // NOTE: this should only be the case if we are connecting to a single server
+    poolOptions.reconnect = true;
+
+    this.s.pool = new Pool(this, poolOptions);
+
+    // setup listeners
+    this.s.pool.on('connect', connectEventHandler(this));
+    this.s.pool.on('close', errorEventHandler(this));
+    this.s.pool.on('error', errorEventHandler(this));
+    this.s.pool.on('parseError', parseErrorEventHandler(this));
+
+    // it is unclear whether consumers should even know about these events
+    // this.s.pool.on('timeout', timeoutEventHandler(this));
+    // this.s.pool.on('reconnect', reconnectEventHandler(this));
+    // this.s.pool.on('reconnectFailed', errorEventHandler(this));
+
+    // relay all command monitoring events
+    relayEvents(this.s.pool, this, ['commandStarted', 'commandSucceeded', 'commandFailed']);
+
+    this.s.state = STATE_CONNECTING;
+
+    // If auth settings have been provided, use them
+    if (options.auth) {
+      this.s.pool.connect.apply(this.s.pool, options.auth);
+      return;
+    }
+
+    this.s.pool.connect();
+  }
+
+  /**
+   * Destroy the server connection
+   *
+   * @param {Boolean} [options.force=false] Force destroy the pool
+   */
+  destroy(options, callback) {
+    if (typeof options === 'function') (callback = options), (options = {});
+    options = Object.assign({}, { force: false }, options);
+
+    const done = err => {
+      this.emit('closed');
+      this.s.state = STATE_DISCONNECTED;
+      if (typeof callback === 'function') {
+        callback(err, null);
+      }
+    };
+
+    if (!this.s.pool) {
+      return done();
+    }
+
+    ['close', 'error', 'timeout', 'parseError', 'connect'].forEach(event => {
+      this.s.pool.removeAllListeners(event);
+    });
+
+    if (this.s.monitorId) {
+      clearTimeout(this.s.monitorId);
+    }
+
+    this.s.pool.destroy(options.force, done);
+  }
+
+  /**
+   * Immediately schedule monitoring of this server. If there already an attempt being made
+   * this will be a no-op.
+   */
+  monitor(options) {
+    options = options || {};
+    if (this.s.state !== STATE_CONNECTED || this.s.monitoring) return;
+    if (this.s.monitorId) clearTimeout(this.s.monitorId);
+    this.s.monitorFunction(this, options);
+  }
+
+  /**
+   * Execute a command
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cmd The command hash
+   * @param {ReadPreference} [options.readPreference] Specify read preference if command supports it
+   * @param {Boolean} [options.serializeFunctions=false] Specify if functions on an object should be serialized.
+   * @param {Boolean} [options.checkKeys=false] Specify if the bson parser should validate keys.
+   * @param {Boolean} [options.ignoreUndefined=false] Specify if the BSON serializer should ignore undefined fields.
+   * @param {Boolean} [options.fullResult=false] Return the full envelope instead of just the result document.
+   * @param {ClientSession} [options.session=null] Session to use for the operation
+   * @param {opResultCallback} callback A callback function
+   */
+  command(ns, cmd, options, callback) {
+    if (typeof options === 'function') {
+      (callback = options), (options = {}), (options = options || {});
+    }
+
+    const error = basicReadValidations(this, options);
+    if (error) {
+      return callback(error, null);
+    }
+
+    // Clone the options
+    options = Object.assign({}, options, { wireProtocolCommand: false });
+
+    // Debug log
+    if (this.s.logger.isDebug()) {
+      this.s.logger.debug(
+        `executing command [${JSON.stringify({
+          ns,
+          cmd,
+          options: debugOptions(DEBUG_FIELDS, options)
+        })}] against ${this.name}`
+      );
+    }
+
+    // error if collation not supported
+    if (collationNotSupported(this, cmd)) {
+      callback(new MongoError(`server ${this.name} does not support collation`));
+      return;
+    }
+
+    wireProtocol.command(this, ns, cmd, options, (err, result) => {
+      if (err) {
+        if (options.session && err instanceof MongoNetworkError) {
+          options.session.serverSession.isDirty = true;
+        }
+
+        if (isSDAMUnrecoverableError(err, this)) {
+          this.emit('error', err);
+        }
+      }
+
+      callback(err, result);
+    });
+  }
+
+  /**
+   * Execute a query against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cmd The command document for the query
+   * @param {object} options Optional settings
+   * @param {function} callback
+   */
+  query(ns, cmd, cursorState, options, callback) {
+    wireProtocol.query(this, ns, cmd, cursorState, options, (err, result) => {
+      if (err) {
+        if (options.session && err instanceof MongoNetworkError) {
+          options.session.serverSession.isDirty = true;
+        }
+
+        if (isSDAMUnrecoverableError(err, this)) {
+          this.emit('error', err);
+        }
+      }
+
+      callback(err, result);
+    });
+  }
+
+  /**
+   * Execute a `getMore` against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cursorState State data associated with the cursor calling this method
+   * @param {object} options Optional settings
+   * @param {function} callback
+   */
+  getMore(ns, cursorState, batchSize, options, callback) {
+    wireProtocol.getMore(this, ns, cursorState, batchSize, options, (err, result) => {
+      if (err) {
+        if (options.session && err instanceof MongoNetworkError) {
+          options.session.serverSession.isDirty = true;
+        }
+
+        if (isSDAMUnrecoverableError(err, this)) {
+          this.emit('error', err);
+        }
+      }
+
+      callback(err, result);
+    });
+  }
+
+  /**
+   * Execute a `killCursors` command against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cursorState State data associated with the cursor calling this method
+   * @param {function} callback
+   */
+  killCursors(ns, cursorState, callback) {
+    wireProtocol.killCursors(this, ns, cursorState, (err, result) => {
+      if (err && isSDAMUnrecoverableError(err, this)) {
+        this.emit('error', err);
+      }
+
+      if (typeof callback === 'function') {
+        callback(err, result);
+      }
+    });
+  }
+
+  /**
+   * Insert one or more documents
+   * @method
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {array} ops An array of documents to insert
+   * @param {boolean} [options.ordered=true] Execute in order or out of order
+   * @param {object} [options.writeConcern={}] Write concern for the operation
+   * @param {Boolean} [options.serializeFunctions=false] Specify if functions on an object should be serialized.
+   * @param {Boolean} [options.ignoreUndefined=false] Specify if the BSON serializer should ignore undefined fields.
+   * @param {ClientSession} [options.session=null] Session to use for the operation
+   * @param {opResultCallback} callback A callback function
+   */
+  insert(ns, ops, options, callback) {
+    executeWriteOperation({ server: this, op: 'insert', ns, ops }, options, callback);
+  }
+
+  /**
+   * Perform one or more update operations
+   * @method
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {array} ops An array of updates
+   * @param {boolean} [options.ordered=true] Execute in order or out of order
+   * @param {object} [options.writeConcern={}] Write concern for the operation
+   * @param {Boolean} [options.serializeFunctions=false] Specify if functions on an object should be serialized.
+   * @param {Boolean} [options.ignoreUndefined=false] Specify if the BSON serializer should ignore undefined fields.
+   * @param {ClientSession} [options.session=null] Session to use for the operation
+   * @param {opResultCallback} callback A callback function
+   */
+  update(ns, ops, options, callback) {
+    executeWriteOperation({ server: this, op: 'update', ns, ops }, options, callback);
+  }
+
+  /**
+   * Perform one or more remove operations
+   * @method
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {array} ops An array of removes
+   * @param {boolean} [options.ordered=true] Execute in order or out of order
+   * @param {object} [options.writeConcern={}] Write concern for the operation
+   * @param {Boolean} [options.serializeFunctions=false] Specify if functions on an object should be serialized.
+   * @param {Boolean} [options.ignoreUndefined=false] Specify if the BSON serializer should ignore undefined fields.
+   * @param {ClientSession} [options.session=null] Session to use for the operation
+   * @param {opResultCallback} callback A callback function
+   */
+  remove(ns, ops, options, callback) {
+    executeWriteOperation({ server: this, op: 'remove', ns, ops }, options, callback);
+  }
+}
+
+Object.defineProperty(Server.prototype, 'clusterTime', {
+  get: function() {
+    return this.s.topology.clusterTime;
+  },
+  set: function(clusterTime) {
+    this.s.topology.clusterTime = clusterTime;
+  }
+});
+
+function basicWriteValidations(server) {
+  if (!server.s.pool) {
+    return new MongoError('server instance is not connected');
+  }
+
+  if (server.s.pool.isDestroyed()) {
+    return new MongoError('server instance pool was destroyed');
+  }
+
+  return null;
+}
+
+function basicReadValidations(server, options) {
+  const error = basicWriteValidations(server, options);
+  if (error) {
+    return error;
+  }
+
+  if (options.readPreference && !(options.readPreference instanceof ReadPreference)) {
+    return new MongoError('readPreference must be an instance of ReadPreference');
+  }
+}
+
+function executeWriteOperation(args, options, callback) {
+  if (typeof options === 'function') (callback = options), (options = {});
+  options = options || {};
+
+  // TODO: once we drop Node 4, use destructuring either here or in arguments.
+  const server = args.server;
+  const op = args.op;
+  const ns = args.ns;
+  const ops = Array.isArray(args.ops) ? args.ops : [args.ops];
+
+  const error = basicWriteValidations(server, options);
+  if (error) {
+    callback(error, null);
+    return;
+  }
+
+  if (collationNotSupported(server, options)) {
+    callback(new MongoError(`server ${server.name} does not support collation`));
+    return;
+  }
+
+  return wireProtocol[op](server, ns, ops, options, (err, result) => {
+    if (err) {
+      if (options.session && err instanceof MongoNetworkError) {
+        options.session.serverSession.isDirty = true;
+      }
+
+      if (isSDAMUnrecoverableError(err, server)) {
+        server.emit('error', err);
+      }
+    }
+
+    callback(err, result);
+  });
+}
+
+function connectEventHandler(server) {
+  return function(pool, conn) {
+    const ismaster = conn.ismaster;
+    server.s.lastIsMasterMS = conn.lastIsMasterMS;
+    if (conn.agreedCompressor) {
+      server.s.pool.options.agreedCompressor = conn.agreedCompressor;
+    }
+
+    if (conn.zlibCompressionLevel) {
+      server.s.pool.options.zlibCompressionLevel = conn.zlibCompressionLevel;
+    }
+
+    if (conn.ismaster.$clusterTime) {
+      const $clusterTime = conn.ismaster.$clusterTime;
+      server.s.sclusterTime = $clusterTime;
+    }
+
+    // log the connection event if requested
+    if (server.s.logger.isInfo()) {
+      server.s.logger.info(
+        `server ${server.name} connected with ismaster [${JSON.stringify(ismaster)}]`
+      );
+    }
+
+    // emit an event indicating that our description has changed
+    server.emit('descriptionReceived', new ServerDescription(server.description.address, ismaster));
+
+    // we are connected and handshaked (guaranteed by the pool)
+    server.s.state = STATE_CONNECTED;
+    server.emit('connect', server);
+  };
+}
+
+function errorEventHandler(server) {
+  return function(err) {
+    if (err) {
+      server.emit('error', new MongoNetworkError(err));
+    }
+
+    server.emit('close');
+  };
+}
+
+function parseErrorEventHandler(server) {
+  return function(err) {
+    server.s.state = STATE_DISCONNECTED;
+    server.emit('error', new MongoParseError(err));
+  };
+}
+
+module.exports = Server;

node_modules/mongodb/lib/core/sdam/server_description.js

@@ -0,0 +1,163 @@
+'use strict';
+
+// An enumeration of server types we know about
+const ServerType = {
+  Standalone: 'Standalone',
+  Mongos: 'Mongos',
+  PossiblePrimary: 'PossiblePrimary',
+  RSPrimary: 'RSPrimary',
+  RSSecondary: 'RSSecondary',
+  RSArbiter: 'RSArbiter',
+  RSOther: 'RSOther',
+  RSGhost: 'RSGhost',
+  Unknown: 'Unknown'
+};
+
+const WRITABLE_SERVER_TYPES = new Set([
+  ServerType.RSPrimary,
+  ServerType.Standalone,
+  ServerType.Mongos
+]);
+
+const DATA_BEARING_SERVER_TYPES = new Set([
+  ServerType.RSPrimary,
+  ServerType.RSSecondary,
+  ServerType.Mongos,
+  ServerType.Standalone
+]);
+
+const ISMASTER_FIELDS = [
+  'minWireVersion',
+  'maxWireVersion',
+  'maxBsonObjectSize',
+  'maxMessageSizeBytes',
+  'maxWriteBatchSize',
+  'compression',
+  'me',
+  'hosts',
+  'passives',
+  'arbiters',
+  'tags',
+  'setName',
+  'setVersion',
+  'electionId',
+  'primary',
+  'logicalSessionTimeoutMinutes',
+  'saslSupportedMechs',
+  '__nodejs_mock_server__',
+  '$clusterTime'
+];
+
+/**
+ * The client's view of a single server, based on the most recent ismaster outcome.
+ *
+ * Internal type, not meant to be directly instantiated
+ */
+class ServerDescription {
+  /**
+   * Create a ServerDescription
+   * @param {String} address The address of the server
+   * @param {Object} [ismaster] An optional ismaster response for this server
+   * @param {Object} [options] Optional settings
+   * @param {Number} [options.roundTripTime] The round trip time to ping this server (in ms)
+   */
+  constructor(address, ismaster, options) {
+    options = options || {};
+    ismaster = Object.assign(
+      {
+        minWireVersion: 0,
+        maxWireVersion: 0,
+        hosts: [],
+        passives: [],
+        arbiters: [],
+        tags: []
+      },
+      ismaster
+    );
+
+    this.address = address;
+    this.error = options.error || null;
+    this.roundTripTime = options.roundTripTime || 0;
+    this.lastUpdateTime = Date.now();
+    this.lastWriteDate = ismaster.lastWrite ? ismaster.lastWrite.lastWriteDate : null;
+    this.opTime = ismaster.lastWrite ? ismaster.lastWrite.opTime : null;
+    this.type = parseServerType(ismaster);
+
+    // direct mappings
+    ISMASTER_FIELDS.forEach(field => {
+      if (typeof ismaster[field] !== 'undefined') this[field] = ismaster[field];
+    });
+
+    // normalize case for hosts
+    if (this.me) this.me = this.me.toLowerCase();
+    this.hosts = this.hosts.map(host => host.toLowerCase());
+    this.passives = this.passives.map(host => host.toLowerCase());
+    this.arbiters = this.arbiters.map(host => host.toLowerCase());
+  }
+
+  get allHosts() {
+    return this.hosts.concat(this.arbiters).concat(this.passives);
+  }
+
+  /**
+   * @return {Boolean} Is this server available for reads
+   */
+  get isReadable() {
+    return this.type === ServerType.RSSecondary || this.isWritable;
+  }
+
+  /**
+   * @return {Boolean} Is this server data bearing
+   */
+  get isDataBearing() {
+    return DATA_BEARING_SERVER_TYPES.has(this.type);
+  }
+
+  /**
+   * @return {Boolean} Is this server available for writes
+   */
+  get isWritable() {
+    return WRITABLE_SERVER_TYPES.has(this.type);
+  }
+}
+
+/**
+ * Parses an `ismaster` message and determines the server type
+ *
+ * @param {Object} ismaster The `ismaster` message to parse
+ * @return {ServerType}
+ */
+function parseServerType(ismaster) {
+  if (!ismaster || !ismaster.ok) {
+    return ServerType.Unknown;
+  }
+
+  if (ismaster.isreplicaset) {
+    return ServerType.RSGhost;
+  }
+
+  if (ismaster.msg && ismaster.msg === 'isdbgrid') {
+    return ServerType.Mongos;
+  }
+
+  if (ismaster.setName) {
+    if (ismaster.hidden) {
+      return ServerType.RSOther;
+    } else if (ismaster.ismaster) {
+      return ServerType.RSPrimary;
+    } else if (ismaster.secondary) {
+      return ServerType.RSSecondary;
+    } else if (ismaster.arbiterOnly) {
+      return ServerType.RSArbiter;
+    } else {
+      return ServerType.RSOther;
+    }
+  }
+
+  return ServerType.Standalone;
+}
+
+module.exports = {
+  ServerDescription,
+  ServerType
+};

node_modules/mongodb/lib/core/sdam/server_selectors.js

@@ -0,0 +1,244 @@
+'use strict';
+const ServerType = require('./server_description').ServerType;
+const TopologyType = require('./topology_description').TopologyType;
+const ReadPreference = require('../topologies/read_preference');
+const MongoError = require('../error').MongoError;
+
+// max staleness constants
+const IDLE_WRITE_PERIOD = 10000;
+const SMALLEST_MAX_STALENESS_SECONDS = 90;
+
+/**
+ * Returns a server selector that selects for writable servers
+ */
+function writableServerSelector() {
+  return function(topologyDescription, servers) {
+    return latencyWindowReducer(topologyDescription, servers.filter(s => s.isWritable));
+  };
+}
+
+/**
+ * Reduces the passed in array of servers by the rules of the "Max Staleness" specification
+ * found here: https://github.com/mongodb/specifications/blob/master/source/max-staleness/max-staleness.rst
+ *
+ * @param {ReadPreference} readPreference The read preference providing max staleness guidance
+ * @param {topologyDescription} topologyDescription The topology description
+ * @param {ServerDescription[]} servers The list of server descriptions to be reduced
+ * @return {ServerDescription[]} The list of servers that satisfy the requirements of max staleness
+ */
+function maxStalenessReducer(readPreference, topologyDescription, servers) {
+  if (readPreference.maxStalenessSeconds == null || readPreference.maxStalenessSeconds < 0) {
+    return servers;
+  }
+
+  const maxStaleness = readPreference.maxStalenessSeconds;
+  const maxStalenessVariance =
+    (topologyDescription.heartbeatFrequencyMS + IDLE_WRITE_PERIOD) / 1000;
+  if (maxStaleness < maxStalenessVariance) {
+    throw new MongoError(`maxStalenessSeconds must be at least ${maxStalenessVariance} seconds`);
+  }
+
+  if (maxStaleness < SMALLEST_MAX_STALENESS_SECONDS) {
+    throw new MongoError(
+      `maxStalenessSeconds must be at least ${SMALLEST_MAX_STALENESS_SECONDS} seconds`
+    );
+  }
+
+  if (topologyDescription.type === TopologyType.ReplicaSetWithPrimary) {
+    const primary = servers.filter(primaryFilter)[0];
+    return servers.reduce((result, server) => {
+      const stalenessMS =
+        server.lastUpdateTime -
+        server.lastWriteDate -
+        (primary.lastUpdateTime - primary.lastWriteDate) +
+        topologyDescription.heartbeatFrequencyMS;
+
+      const staleness = stalenessMS / 1000;
+      if (staleness <= readPreference.maxStalenessSeconds) result.push(server);
+      return result;
+    }, []);
+  } else if (topologyDescription.type === TopologyType.ReplicaSetNoPrimary) {
+    const sMax = servers.reduce((max, s) => (s.lastWriteDate > max.lastWriteDate ? s : max));
+    return servers.reduce((result, server) => {
+      const stalenessMS =
+        sMax.lastWriteDate - server.lastWriteDate + topologyDescription.heartbeatFrequencyMS;
+
+      const staleness = stalenessMS / 1000;
+      if (staleness <= readPreference.maxStalenessSeconds) result.push(server);
+      return result;
+    }, []);
+  }
+
+  return servers;
+}
+
+/**
+ * Determines whether a server's tags match a given set of tags
+ *
+ * @param {String[]} tagSet The requested tag set to match
+ * @param {String[]} serverTags The server's tags
+ */
+function tagSetMatch(tagSet, serverTags) {
+  const keys = Object.keys(tagSet);
+  const serverTagKeys = Object.keys(serverTags);
+  for (let i = 0; i < keys.length; ++i) {
+    const key = keys[i];
+    if (serverTagKeys.indexOf(key) === -1 || serverTags[key] !== tagSet[key]) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+/**
+ * Reduces a set of server descriptions based on tags requested by the read preference
+ *
+ * @param {ReadPreference} readPreference The read preference providing the requested tags
+ * @param {ServerDescription[]} servers The list of server descriptions to reduce
+ * @return {ServerDescription[]} The list of servers matching the requested tags
+ */
+function tagSetReducer(readPreference, servers) {
+  if (
+    readPreference.tags == null ||
+    (Array.isArray(readPreference.tags) && readPreference.tags.length === 0)
+  ) {
+    return servers;
+  }
+
+  for (let i = 0; i < readPreference.tags.length; ++i) {
+    const tagSet = readPreference.tags[i];
+    const serversMatchingTagset = servers.reduce((matched, server) => {
+      if (tagSetMatch(tagSet, server.tags)) matched.push(server);
+      return matched;
+    }, []);
+
+    if (serversMatchingTagset.length) {
+      return serversMatchingTagset;
+    }
+  }
+
+  return [];
+}
+
+/**
+ * Reduces a list of servers to ensure they fall within an acceptable latency window. This is
+ * further specified in the "Server Selection" specification, found here:
+ * https://github.com/mongodb/specifications/blob/master/source/server-selection/server-selection.rst
+ *
+ * @param {topologyDescription} topologyDescription The topology description
+ * @param {ServerDescription[]} servers The list of servers to reduce
+ * @returns {ServerDescription[]} The servers which fall within an acceptable latency window
+ */
+function latencyWindowReducer(topologyDescription, servers) {
+  const low = servers.reduce(
+    (min, server) => (min === -1 ? server.roundTripTime : Math.min(server.roundTripTime, min)),
+    -1
+  );
+
+  const high = low + topologyDescription.localThresholdMS;
+
+  return servers.reduce((result, server) => {
+    if (server.roundTripTime <= high && server.roundTripTime >= low) result.push(server);
+    return result;
+  }, []);
+}
+
+// filters
+function primaryFilter(server) {
+  return server.type === ServerType.RSPrimary;
+}
+
+function secondaryFilter(server) {
+  return server.type === ServerType.RSSecondary;
+}
+
+function nearestFilter(server) {
+  return server.type === ServerType.RSSecondary || server.type === ServerType.RSPrimary;
+}
+
+function knownFilter(server) {
+  return server.type !== ServerType.Unknown;
+}
+
+/**
+ * Returns a function which selects servers based on a provided read preference
+ *
+ * @param {ReadPreference} readPreference The read preference to select with
+ */
+function readPreferenceServerSelector(readPreference) {
+  if (!readPreference.isValid()) {
+    throw new TypeError('Invalid read preference specified');
+  }
+
+  return function(topologyDescription, servers) {
+    const commonWireVersion = topologyDescription.commonWireVersion;
+    if (
+      commonWireVersion &&
+      (readPreference.minWireVersion && readPreference.minWireVersion > commonWireVersion)
+    ) {
+      throw new MongoError(
+        `Minimum wire version '${
+          readPreference.minWireVersion
+        }' required, but found '${commonWireVersion}'`
+      );
+    }
+
+    if (
+      topologyDescription.type === TopologyType.Single ||
+      topologyDescription.type === TopologyType.Sharded
+    ) {
+      return latencyWindowReducer(topologyDescription, servers.filter(knownFilter));
+    }
+
+    if (readPreference.mode === ReadPreference.PRIMARY) {
+      return servers.filter(primaryFilter);
+    }
+
+    if (readPreference.mode === ReadPreference.SECONDARY) {
+      return latencyWindowReducer(
+        topologyDescription,
+        tagSetReducer(
+          readPreference,
+          maxStalenessReducer(readPreference, topologyDescription, servers)
+        )
+      ).filter(secondaryFilter);
+    } else if (readPreference.mode === ReadPreference.NEAREST) {
+      return latencyWindowReducer(
+        topologyDescription,
+        tagSetReducer(
+          readPreference,
+          maxStalenessReducer(readPreference, topologyDescription, servers)
+        )
+      ).filter(nearestFilter);
+    } else if (readPreference.mode === ReadPreference.SECONDARY_PREFERRED) {
+      const result = latencyWindowReducer(
+        topologyDescription,
+        tagSetReducer(
+          readPreference,
+          maxStalenessReducer(readPreference, topologyDescription, servers)
+        )
+      ).filter(secondaryFilter);
+
+      return result.length === 0 ? servers.filter(primaryFilter) : result;
+    } else if (readPreference.mode === ReadPreference.PRIMARY_PREFERRED) {
+      const result = servers.filter(primaryFilter);
+      if (result.length) {
+        return result;
+      }
+
+      return latencyWindowReducer(
+        topologyDescription,
+        tagSetReducer(
+          readPreference,
+          maxStalenessReducer(readPreference, topologyDescription, servers)
+        )
+      ).filter(secondaryFilter);
+    }
+  };
+}
+
+module.exports = {
+  writableServerSelector,
+  readPreferenceServerSelector
+};

node_modules/mongodb/lib/core/sdam/srv_polling.js

@@ -0,0 +1,135 @@
+'use strict';
+
+const Logger = require('../connection/logger');
+const EventEmitter = require('events').EventEmitter;
+const dns = require('dns');
+/**
+ * Determines whether a provided address matches the provided parent domain in order
+ * to avoid certain attack vectors.
+ *
+ * @param {String} srvAddress The address to check against a domain
+ * @param {String} parentDomain The domain to check the provided address against
+ * @return {Boolean} Whether the provided address matches the parent domain
+ */
+function matchesParentDomain(srvAddress, parentDomain) {
+  const regex = /^.*?\./;
+  const srv = `.${srvAddress.replace(regex, '')}`;
+  const parent = `.${parentDomain.replace(regex, '')}`;
+  return srv.endsWith(parent);
+}
+
+class SrvPollingEvent {
+  constructor(srvRecords) {
+    this.srvRecords = srvRecords;
+  }
+
+  addresses() {
+    return new Set(this.srvRecords.map(record => `${record.name}:${record.port}`));
+  }
+}
+
+class SrvPoller extends EventEmitter {
+  /**
+   * @param {object} options
+   * @param {string} options.srvHost
+   * @param {number} [options.heartbeatFrequencyMS]
+   * @param {function} [options.logger]
+   * @param {string} [options.loggerLevel]
+   */
+  constructor(options) {
+    super();
+
+    if (!options || !options.srvHost) {
+      throw new TypeError('options for SrvPoller must exist and include srvHost');
+    }
+
+    this.srvHost = options.srvHost;
+    this.rescanSrvIntervalMS = 60000;
+    this.heartbeatFrequencyMS = options.heartbeatFrequencyMS || 10000;
+    this.logger = Logger('srvPoller', options);
+
+    this.haMode = false;
+    this.generation = 0;
+
+    this._timeout = null;
+  }
+
+  get srvAddress() {
+    return `_mongodb._tcp.${this.srvHost}`;
+  }
+
+  get intervalMS() {
+    return this.haMode ? this.heartbeatFrequencyMS : this.rescanSrvIntervalMs;
+  }
+
+  start() {
+    if (!this._timeout) {
+      this.schedule();
+    }
+  }
+
+  stop() {
+    if (this._timeout) {
+      clearTimeout(this._timeout);
+      this.generation += 1;
+      this._timeout = null;
+    }
+  }
+
+  schedule() {
+    clearTimeout(this._timeout);
+    this._timeout = setTimeout(() => this._poll(), this.intervalMS);
+  }
+
+  success(srvRecords) {
+    this.haMode = false;
+    this.schedule();
+    this.emit('srvRecordDiscovery', new SrvPollingEvent(srvRecords));
+  }
+
+  failure(message, obj) {
+    this.logger.warn(message, obj);
+    this.haMode = true;
+    this.schedule();
+  }
+
+  parentDomainMismatch(srvRecord) {
+    this.logger.warn(
+      `parent domain mismatch on SRV record (${srvRecord.name}:${srvRecord.port})`,
+      srvRecord
+    );
+  }
+
+  _poll() {
+    const generation = this.generation;
+    dns.resolveSrv(this.srvAddress, (err, srvRecords) => {
+      if (generation !== this.generation) {
+        return;
+      }
+
+      if (err) {
+        this.failure('DNS error', err);
+        return;
+      }
+
+      const finalAddresses = [];
+      srvRecords.forEach(record => {
+        if (matchesParentDomain(record.name, this.srvHost)) {
+          finalAddresses.push(record);
+        } else {
+          this.parentDomainMismatch(record);
+        }
+      });
+
+      if (!finalAddresses.length) {
+        this.failure('No valid addresses found at host');
+        return;
+      }
+
+      this.success(finalAddresses);
+    });
+  }
+}
+
+module.exports.SrvPollingEvent = SrvPollingEvent;
+module.exports.SrvPoller = SrvPoller;

node_modules/mongodb/lib/core/sdam/topology_description.js

@@ -0,0 +1,408 @@
+'use strict';
+const ServerType = require('./server_description').ServerType;
+const ServerDescription = require('./server_description').ServerDescription;
+const WIRE_CONSTANTS = require('../wireprotocol/constants');
+
+// contstants related to compatability checks
+const MIN_SUPPORTED_SERVER_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_SERVER_VERSION;
+const MAX_SUPPORTED_SERVER_VERSION = WIRE_CONSTANTS.MAX_SUPPORTED_SERVER_VERSION;
+const MIN_SUPPORTED_WIRE_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_WIRE_VERSION;
+const MAX_SUPPORTED_WIRE_VERSION = WIRE_CONSTANTS.MAX_SUPPORTED_WIRE_VERSION;
+
+// An enumeration of topology types we know about
+const TopologyType = {
+  Single: 'Single',
+  ReplicaSetNoPrimary: 'ReplicaSetNoPrimary',
+  ReplicaSetWithPrimary: 'ReplicaSetWithPrimary',
+  Sharded: 'Sharded',
+  Unknown: 'Unknown'
+};
+
+// Representation of a deployment of servers
+class TopologyDescription {
+  /**
+   * Create a TopologyDescription
+   *
+   * @param {string} topologyType
+   * @param {Map<string, ServerDescription>} serverDescriptions the a map of address to ServerDescription
+   * @param {string} setName
+   * @param {number} maxSetVersion
+   * @param {ObjectId} maxElectionId
+   */
+  constructor(
+    topologyType,
+    serverDescriptions,
+    setName,
+    maxSetVersion,
+    maxElectionId,
+    commonWireVersion,
+    options,
+    error
+  ) {
+    options = options || {};
+
+    // TODO: consider assigning all these values to a temporary value `s` which
+    //       we use `Object.freeze` on, ensuring the internal state of this type
+    //       is immutable.
+    this.type = topologyType || TopologyType.Unknown;
+    this.setName = setName || null;
+    this.maxSetVersion = maxSetVersion || null;
+    this.maxElectionId = maxElectionId || null;
+    this.servers = serverDescriptions || new Map();
+    this.stale = false;
+    this.compatible = true;
+    this.compatibilityError = null;
+    this.logicalSessionTimeoutMinutes = null;
+    this.heartbeatFrequencyMS = options.heartbeatFrequencyMS || 0;
+    this.localThresholdMS = options.localThresholdMS || 0;
+    this.options = options;
+    this.error = error;
+    this.commonWireVersion = commonWireVersion || null;
+
+    // determine server compatibility
+    for (const serverDescription of this.servers.values()) {
+      if (serverDescription.type === ServerType.Unknown) continue;
+
+      if (serverDescription.minWireVersion > MAX_SUPPORTED_WIRE_VERSION) {
+        this.compatible = false;
+        this.compatibilityError = `Server at ${serverDescription.address} requires wire version ${
+          serverDescription.minWireVersion
+        }, but this version of the driver only supports up to ${MAX_SUPPORTED_WIRE_VERSION} (MongoDB ${MAX_SUPPORTED_SERVER_VERSION})`;
+      }
+
+      if (serverDescription.maxWireVersion < MIN_SUPPORTED_WIRE_VERSION) {
+        this.compatible = false;
+        this.compatibilityError = `Server at ${serverDescription.address} reports wire version ${
+          serverDescription.maxWireVersion
+        }, but this version of the driver requires at least ${MIN_SUPPORTED_WIRE_VERSION} (MongoDB ${MIN_SUPPORTED_SERVER_VERSION}).`;
+        break;
+      }
+    }
+
+    // Whenever a client updates the TopologyDescription from an ismaster response, it MUST set
+    // TopologyDescription.logicalSessionTimeoutMinutes to the smallest logicalSessionTimeoutMinutes
+    // value among ServerDescriptions of all data-bearing server types. If any have a null
+    // logicalSessionTimeoutMinutes, then TopologyDescription.logicalSessionTimeoutMinutes MUST be
+    // set to null.
+    const readableServers = Array.from(this.servers.values()).filter(s => s.isReadable);
+    this.logicalSessionTimeoutMinutes = readableServers.reduce((result, server) => {
+      if (server.logicalSessionTimeoutMinutes == null) return null;
+      if (result == null) return server.logicalSessionTimeoutMinutes;
+      return Math.min(result, server.logicalSessionTimeoutMinutes);
+    }, null);
+  }
+
+  /**
+   * Returns a new TopologyDescription based on the SrvPollingEvent
+   * @param {SrvPollingEvent} ev The event
+   */
+  updateFromSrvPollingEvent(ev) {
+    const newAddresses = ev.addresses();
+    const serverDescriptions = new Map(this.servers);
+    for (const server of this.servers) {
+      if (newAddresses.has(server[0])) {
+        newAddresses.delete(server[0]);
+      } else {
+        serverDescriptions.delete(server[0]);
+      }
+    }
+
+    if (serverDescriptions.size === this.servers.size && newAddresses.size === 0) {
+      return this;
+    }
+
+    for (const address of newAddresses) {
+      serverDescriptions.set(address, new ServerDescription(address));
+    }
+
+    return new TopologyDescription(
+      this.type,
+      serverDescriptions,
+      this.setName,
+      this.maxSetVersion,
+      this.maxElectionId,
+      this.commonWireVersion,
+      this.options,
+      null
+    );
+  }
+
+  /**
+   * Returns a copy of this description updated with a given ServerDescription
+   *
+   * @param {ServerDescription} serverDescription
+   */
+  update(serverDescription) {
+    const address = serverDescription.address;
+    // NOTE: there are a number of prime targets for refactoring here
+    //       once we support destructuring assignments
+
+    // potentially mutated values
+    let topologyType = this.type;
+    let setName = this.setName;
+    let maxSetVersion = this.maxSetVersion;
+    let maxElectionId = this.maxElectionId;
+    let commonWireVersion = this.commonWireVersion;
+    let error = serverDescription.error || null;
+
+    const serverType = serverDescription.type;
+    let serverDescriptions = new Map(this.servers);
+
+    // update common wire version
+    if (serverDescription.maxWireVersion !== 0) {
+      if (commonWireVersion == null) {
+        commonWireVersion = serverDescription.maxWireVersion;
+      } else {
+        commonWireVersion = Math.min(commonWireVersion, serverDescription.maxWireVersion);
+      }
+    }
+
+    // update the actual server description
+    serverDescriptions.set(address, serverDescription);
+
+    if (topologyType === TopologyType.Single) {
+      // once we are defined as single, that never changes
+      return new TopologyDescription(
+        TopologyType.Single,
+        serverDescriptions,
+        setName,
+        maxSetVersion,
+        maxElectionId,
+        commonWireVersion,
+        this.options,
+        error
+      );
+    }
+
+    if (topologyType === TopologyType.Unknown) {
+      if (serverType === ServerType.Standalone) {
+        serverDescriptions.delete(address);
+      } else {
+        topologyType = topologyTypeForServerType(serverType);
+      }
+    }
+
+    if (topologyType === TopologyType.Sharded) {
+      if ([ServerType.Mongos, ServerType.Unknown].indexOf(serverType) === -1) {
+        serverDescriptions.delete(address);
+      }
+    }
+
+    if (topologyType === TopologyType.ReplicaSetNoPrimary) {
+      if ([ServerType.Mongos, ServerType.Unknown].indexOf(serverType) >= 0) {
+        serverDescriptions.delete(address);
+      }
+
+      if (serverType === ServerType.RSPrimary) {
+        const result = updateRsFromPrimary(
+          serverDescriptions,
+          setName,
+          serverDescription,
+          maxSetVersion,
+          maxElectionId
+        );
+
+        (topologyType = result[0]),
+          (setName = result[1]),
+          (maxSetVersion = result[2]),
+          (maxElectionId = result[3]);
+      } else if (
+        [ServerType.RSSecondary, ServerType.RSArbiter, ServerType.RSOther].indexOf(serverType) >= 0
+      ) {
+        const result = updateRsNoPrimaryFromMember(serverDescriptions, setName, serverDescription);
+        (topologyType = result[0]), (setName = result[1]);
+      }
+    }
+
+    if (topologyType === TopologyType.ReplicaSetWithPrimary) {
+      if ([ServerType.Standalone, ServerType.Mongos].indexOf(serverType) >= 0) {
+        serverDescriptions.delete(address);
+        topologyType = checkHasPrimary(serverDescriptions);
+      } else if (serverType === ServerType.RSPrimary) {
+        const result = updateRsFromPrimary(
+          serverDescriptions,
+          setName,
+          serverDescription,
+          maxSetVersion,
+          maxElectionId
+        );
+
+        (topologyType = result[0]),
+          (setName = result[1]),
+          (maxSetVersion = result[2]),
+          (maxElectionId = result[3]);
+      } else if (
+        [ServerType.RSSecondary, ServerType.RSArbiter, ServerType.RSOther].indexOf(serverType) >= 0
+      ) {
+        topologyType = updateRsWithPrimaryFromMember(
+          serverDescriptions,
+          setName,
+          serverDescription
+        );
+      } else {
+        topologyType = checkHasPrimary(serverDescriptions);
+      }
+    }
+
+    return new TopologyDescription(
+      topologyType,
+      serverDescriptions,
+      setName,
+      maxSetVersion,
+      maxElectionId,
+      commonWireVersion,
+      this.options,
+      error
+    );
+  }
+
+  /**
+   * Determines if the topology description has any known servers
+   */
+  get hasKnownServers() {
+    return Array.from(this.servers.values()).some(sd => sd.type !== ServerDescription.Unknown);
+  }
+
+  /**
+   * Determines if this topology description has a data-bearing server available.
+   */
+  get hasDataBearingServers() {
+    return Array.from(this.servers.values()).some(sd => sd.isDataBearing);
+  }
+
+  /**
+   * Determines if the topology has a definition for the provided address
+   *
+   * @param {String} address
+   * @return {Boolean} Whether the topology knows about this server
+   */
+  hasServer(address) {
+    return this.servers.has(address);
+  }
+}
+
+function topologyTypeForServerType(serverType) {
+  if (serverType === ServerType.Mongos) return TopologyType.Sharded;
+  if (serverType === ServerType.RSPrimary) return TopologyType.ReplicaSetWithPrimary;
+  return TopologyType.ReplicaSetNoPrimary;
+}
+
+function updateRsFromPrimary(
+  serverDescriptions,
+  setName,
+  serverDescription,
+  maxSetVersion,
+  maxElectionId
+) {
+  setName = setName || serverDescription.setName;
+  if (setName !== serverDescription.setName) {
+    serverDescriptions.delete(serverDescription.address);
+    return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];
+  }
+
+  const electionIdOID = serverDescription.electionId ? serverDescription.electionId.$oid : null;
+  const maxElectionIdOID = maxElectionId ? maxElectionId.$oid : null;
+  if (serverDescription.setVersion != null && electionIdOID != null) {
+    if (maxSetVersion != null && maxElectionIdOID != null) {
+      if (maxSetVersion > serverDescription.setVersion || maxElectionIdOID > electionIdOID) {
+        // this primary is stale, we must remove it
+        serverDescriptions.set(
+          serverDescription.address,
+          new ServerDescription(serverDescription.address)
+        );
+
+        return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];
+      }
+    }
+
+    maxElectionId = serverDescription.electionId;
+  }
+
+  if (
+    serverDescription.setVersion != null &&
+    (maxSetVersion == null || serverDescription.setVersion > maxSetVersion)
+  ) {
+    maxSetVersion = serverDescription.setVersion;
+  }
+
+  // We've heard from the primary. Is it the same primary as before?
+  for (const address of serverDescriptions.keys()) {
+    const server = serverDescriptions.get(address);
+
+    if (server.type === ServerType.RSPrimary && server.address !== serverDescription.address) {
+      // Reset old primary's type to Unknown.
+      serverDescriptions.set(address, new ServerDescription(server.address));
+
+      // There can only be one primary
+      break;
+    }
+  }
+
+  // Discover new hosts from this primary's response.
+  serverDescription.allHosts.forEach(address => {
+    if (!serverDescriptions.has(address)) {
+      serverDescriptions.set(address, new ServerDescription(address));
+    }
+  });
+
+  // Remove hosts not in the response.
+  const currentAddresses = Array.from(serverDescriptions.keys());
+  const responseAddresses = serverDescription.allHosts;
+  currentAddresses.filter(addr => responseAddresses.indexOf(addr) === -1).forEach(address => {
+    serverDescriptions.delete(address);
+  });
+
+  return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];
+}
+
+function updateRsWithPrimaryFromMember(serverDescriptions, setName, serverDescription) {
+  if (setName == null) {
+    throw new TypeError('setName is required');
+  }
+
+  if (
+    setName !== serverDescription.setName ||
+    (serverDescription.me && serverDescription.address !== serverDescription.me)
+  ) {
+    serverDescriptions.delete(serverDescription.address);
+  }
+
+  return checkHasPrimary(serverDescriptions);
+}
+
+function updateRsNoPrimaryFromMember(serverDescriptions, setName, serverDescription) {
+  let topologyType = TopologyType.ReplicaSetNoPrimary;
+
+  setName = setName || serverDescription.setName;
+  if (setName !== serverDescription.setName) {
+    serverDescriptions.delete(serverDescription.address);
+    return [topologyType, setName];
+  }
+
+  serverDescription.allHosts.forEach(address => {
+    if (!serverDescriptions.has(address)) {
+      serverDescriptions.set(address, new ServerDescription(address));
+    }
+  });
+
+  if (serverDescription.me && serverDescription.address !== serverDescription.me) {
+    serverDescriptions.delete(serverDescription.address);
+  }
+
+  return [topologyType, setName];
+}
+
+function checkHasPrimary(serverDescriptions) {
+  for (const addr of serverDescriptions.keys()) {
+    if (serverDescriptions.get(addr).type === ServerType.RSPrimary) {
+      return TopologyType.ReplicaSetWithPrimary;
+    }
+  }
+
+  return TopologyType.ReplicaSetNoPrimary;
+}
+
+module.exports = {
+  TopologyType,
+  TopologyDescription
+};