refactor: group all server selection related code into a module

Author: mbroadst

lib/core/sdam/common.js

@@ -28,11 +28,32 @@ const ServerType = {
   Unknown: 'Unknown'
 };
 
+const TOPOLOGY_DEFAULTS = {
+  useUnifiedTopology: true,
+  localThresholdMS: 15,
+  serverSelectionTimeoutMS: 30000,
+  heartbeatFrequencyMS: 10000,
+  minHeartbeatFrequencyMS: 500
+};
+
+function drainTimerQueue(queue) {
+  queue.forEach(clearTimeout);
+  queue.clear();
+}
+
+function clearAndRemoveTimerFrom(timer, timers) {
+  clearTimeout(timer);
+  return timers.delete(timer);
+}
+
 module.exports = {
   STATE_CLOSING,
   STATE_CLOSED,
   STATE_CONNECTING,
   STATE_CONNECTED,
+  TOPOLOGY_DEFAULTS,
   TopologyType,
-  ServerType
+  ServerType,
+  drainTimerQueue,
+  clearAndRemoveTimerFrom
 };

lib/core/sdam/server_selectors.js renamed to lib/core/sdam/server_selection.js

@@ -3,6 +3,15 @@ const ServerType = require('./common').ServerType;
 const TopologyType = require('./common').TopologyType;
 const ReadPreference = require('../topologies/read_preference');
 const MongoError = require('../error').MongoError;
+const calculateDurationInMs = require('../utils').calculateDurationInMs;
+const MongoTimeoutError = require('../error').MongoTimeoutError;
+
+const common = require('./common');
+const STATE_CONNECTED = common.STATE_CONNECTED;
+const STATE_CONNECTING = common.STATE_CONNECTING;
+const TOPOLOGY_DEFAULTS = common.TOPOLOGY_DEFAULTS;
+const drainTimerQueue = common.drainTimerQueue;
+const clearAndRemoveTimerFrom = common.clearAndRemoveTimerFrom;
 
 // max staleness constants
 const IDLE_WRITE_PERIOD = 10000;
@@ -238,7 +247,117 @@ function readPreferenceServerSelector(readPreference) {
   };
 }
 
+/**
+ * Selects servers using the provided selector
+ *
+ * @private
+ * @param {Topology} topology The topology to select servers from
+ * @param {function} selector The predicate used for selecting servers
+ * @param {Number} timeout The max time we are willing wait for selection
+ * @param {Number} start A high precision timestamp for the start of the selection process
+ * @param {function} callback The callback used to convey errors or the resultant servers
+ */
+function selectServers(topology, selector, timeout, start, callback) {
+  const duration = calculateDurationInMs(start);
+  if (duration >= timeout) {
+    return callback(
+      new MongoTimeoutError(`Server selection timed out after ${timeout} ms`),
+      topology.description.error
+    );
+  }
+
+  // ensure we are connected
+  if (topology.s.state !== STATE_CONNECTED && topology.s.state !== STATE_CONNECTING) {
+    topology.connect();
+
+    // we want to make sure we're still within the requested timeout window
+    const failToConnectTimer = setTimeout(() => {
+      topology.removeListener('connect', connectHandler);
+      callback(
+        new MongoTimeoutError('Server selection timed out waiting to connect'),
+        topology.description.error
+      );
+    }, timeout - duration);
+
+    const connectHandler = () => {
+      clearAndRemoveTimerFrom(failToConnectTimer, topology.s.connectionTimers);
+      selectServers(topology, selector, timeout, process.hrtime(), callback);
+    };
+
+    topology.s.connectionTimers.add(failToConnectTimer);
+    topology.once('connect', connectHandler);
+    return;
+  }
+
+  // otherwise, attempt server selection
+  const serverDescriptions = Array.from(topology.description.servers.values());
+  let descriptions;
+
+  // support server selection by options with readPreference
+  if (typeof selector === 'object') {
+    const readPreference = selector.readPreference
+      ? selector.readPreference
+      : ReadPreference.primary;
+
+    selector = readPreferenceServerSelector(readPreference);
+  }
+
+  try {
+    descriptions = selector
+      ? selector(topology.description, serverDescriptions)
+      : serverDescriptions;
+  } catch (e) {
+    return callback(e, null);
+  }
+
+  if (descriptions.length) {
+    const servers = descriptions.map(description => topology.s.servers.get(description.address));
+    return callback(null, servers);
+  }
+
+  const retrySelection = () => {
+    // clear all existing monitor timers
+    drainTimerQueue(topology.s.monitorTimers);
+
+    // ensure all server monitors attempt monitoring soon
+    topology.s.servers.forEach(server => {
+      const timer = setTimeout(
+        () => server.monitor({ heartbeatFrequencyMS: topology.description.heartbeatFrequencyMS }),
+        TOPOLOGY_DEFAULTS.minHeartbeatFrequencyMS
+      );
+
+      topology.s.monitorTimers.add(timer);
+    });
+
+    const iterationTimer = setTimeout(() => {
+      topology.removeListener('topologyDescriptionChanged', descriptionChangedHandler);
+      callback(
+        new MongoTimeoutError(
+          `Server selection timed out after ${timeout} ms`,
+          topology.description.error
+        )
+      );
+    }, timeout - duration);
+
+    const descriptionChangedHandler = () => {
+      // successful iteration, clear the check timer
+      clearAndRemoveTimerFrom(iterationTimer, topology.s.iterationTimers);
+
+      // topology description has changed due to monitoring, reattempt server selection
+      selectServers(topology, selector, timeout, start, callback);
+    };
+
+    // track this timer in case we need to clean it up outside this loop
+    topology.s.iterationTimers.add(iterationTimer);
+
+    topology.once('topologyDescriptionChanged', descriptionChangedHandler);
+  };
+
+  retrySelection();
+}
+
 module.exports = {
+  selectServers,
   writableServerSelector,
   readPreferenceServerSelector
 };

lib/core/sdam/topology.js

@@ -5,13 +5,9 @@ const ServerType = require('./common').ServerType;
 const TopologyDescription = require('./topology_description');
 const TopologyType = require('./common').TopologyType;
 const monitoring = require('./monitoring');
-const calculateDurationInMs = require('../utils').calculateDurationInMs;
-const MongoTimeoutError = require('../error').MongoTimeoutError;
 const Server = require('./server');
 const relayEvents = require('../utils').relayEvents;
 const ReadPreference = require('../topologies/read_preference');
-const readPreferenceServerSelector = require('./server_selectors').readPreferenceServerSelector;
-const writableServerSelector = require('./server_selectors').writableServerSelector;
 const isRetryableWritesSupported = require('../topologies/shared').isRetryableWritesSupported;
 const CoreCursor = require('../cursor').CoreCursor;
 const deprecate = require('util').deprecate;
@@ -27,20 +23,19 @@ const SrvPoller = require('./srv_polling').SrvPoller;
 const getMMAPError = require('../topologies/shared').getMMAPError;
 const makeStateMachine = require('../utils').makeStateMachine;
 const eachAsync = require('../utils').eachAsync;
+
 const common = require('./common');
+const drainTimerQueue = common.drainTimerQueue;
+const clearAndRemoveTimerFrom = common.clearAndRemoveTimerFrom;
+
+const serverSelection = require('./server_selection');
+const readPreferenceServerSelector = serverSelection.readPreferenceServerSelector;
+const writableServerSelector = serverSelection.writableServerSelector;
+const selectServers = serverSelection.selectServers;
 
 // Global state
 let globalTopologyCounter = 0;
 
-// Constants
-const TOPOLOGY_DEFAULTS = {
-  useUnifiedTopology: true,
-  localThresholdMS: 15,
-  serverSelectionTimeoutMS: 30000,
-  heartbeatFrequencyMS: 10000,
-  minHeartbeatFrequencyMS: 500
-};
-
 // events that we relay to the `Topology`
 const SERVER_RELAY_EVENTS = [
   'serverHeartbeatStarted',
@@ -114,7 +109,7 @@ class Topology extends EventEmitter {
       seedlist = parseStringSeedlist(seedlist);
     }
 
-    options = Object.assign({}, TOPOLOGY_DEFAULTS, options);
+    options = Object.assign({}, common.TOPOLOGY_DEFAULTS, options);
 
     const topologyType = topologyTypeFromSeedlist(seedlist, options);
     const topologyId = globalTopologyCounter++;
@@ -810,115 +805,6 @@ function randomSelection(array) {
   return array[Math.floor(Math.random() * array.length)];
 }
 
-/**
- * Selects servers using the provided selector
- *
- * @private
- * @param {Topology} topology The topology to select servers from
- * @param {function} selector The actual predicate used for selecting servers
- * @param {Number} timeout The max time we are willing wait for selection
- * @param {Number} start A high precision timestamp for the start of the selection process
- * @param {function} callback The callback used to convey errors or the resultant servers
- */
-function selectServers(topology, selector, timeout, start, callback) {
-  const duration = calculateDurationInMs(start);
-  if (duration >= timeout) {
-    return callback(
-      new MongoTimeoutError(`Server selection timed out after ${timeout} ms`),
-      topology.description.error
-    );
-  }
-
-  // ensure we are connected
-  if (topology.s.state !== STATE_CONNECTED && topology.s.state !== STATE_CONNECTING) {
-    topology.connect();
-
-    // we want to make sure we're still within the requested timeout window
-    const failToConnectTimer = setTimeout(() => {
-      topology.removeListener('connect', connectHandler);
-      callback(
-        new MongoTimeoutError('Server selection timed out waiting to connect'),
-        topology.description.error
-      );
-    }, timeout - duration);
-
-    const connectHandler = () => {
-      clearAndRemoveTimerFrom(failToConnectTimer, topology.s.connectionTimers);
-      selectServers(topology, selector, timeout, process.hrtime(), callback);
-    };
-
-    topology.s.connectionTimers.add(failToConnectTimer);
-    topology.once('connect', connectHandler);
-    return;
-  }
-
-  // otherwise, attempt server selection
-  const serverDescriptions = Array.from(topology.description.servers.values());
-  let descriptions;
-
-  // support server selection by options with readPreference
-  if (typeof selector === 'object') {
-    const readPreference = selector.readPreference
-      ? selector.readPreference
-      : ReadPreference.primary;
-
-    selector = readPreferenceServerSelector(readPreference);
-  }
-
-  try {
-    descriptions = selector
-      ? selector(topology.description, serverDescriptions)
-      : serverDescriptions;
-  } catch (e) {
-    return callback(e, null);
-  }
-
-  if (descriptions.length) {
-    const servers = descriptions.map(description => topology.s.servers.get(description.address));
-    return callback(null, servers);
-  }
-
-  const retrySelection = () => {
-    // clear all existing monitor timers
-    drainTimerQueue(topology.s.monitorTimers);
-
-    // ensure all server monitors attempt monitoring soon
-    topology.s.servers.forEach(server => {
-      const timer = setTimeout(
-        () => server.monitor({ heartbeatFrequencyMS: topology.description.heartbeatFrequencyMS }),
-        TOPOLOGY_DEFAULTS.minHeartbeatFrequencyMS
-      );
-
-      topology.s.monitorTimers.add(timer);
-    });
-
-    const iterationTimer = setTimeout(() => {
-      topology.removeListener('topologyDescriptionChanged', descriptionChangedHandler);
-      callback(
-        new MongoTimeoutError(
-          `Server selection timed out after ${timeout} ms`,
-          topology.description.error
-        )
-      );
-    }, timeout - duration);
-
-    const descriptionChangedHandler = () => {
-      // successful iteration, clear the check timer
-      clearAndRemoveTimerFrom(iterationTimer, topology.s.iterationTimers);
-
-      // topology description has changed due to monitoring, reattempt server selection
-      selectServers(topology, selector, timeout, start, callback);
-    };
-
-    // track this timer in case we need to clean it up outside this loop
-    topology.s.iterationTimers.add(iterationTimer);
-
-    topology.once('topologyDescriptionChanged', descriptionChangedHandler);
-  };
-
-  retrySelection();
-}
-
 function createAndConnectServer(topology, serverDescription, connectDelay) {
   topology.emit(
     'serverOpening',
@@ -965,16 +851,6 @@ function resetServer(topology, serverDescription) {
   topology.s.servers.set(serverDescription.address, newServer);
 }
 
-function drainTimerQueue(queue) {
-  queue.forEach(clearTimeout);
-  queue.clear();
-}
-
-function clearAndRemoveTimerFrom(timer, timers) {
-  clearTimeout(timer);
-  return timers.delete(timer);
-}
-
 /**
  * Create `Server` instances for all initially known servers, connect them, and assign
  * them to the passed in `Topology`.

test/unit/sdam/server_selection/spec_tests.js

@@ -10,7 +10,7 @@ const ReadPreference = core.ReadPreference;
 const Server = require('../../../../lib/core/sdam/server');
 const ServerType = require('../../../../lib/core/sdam/common').ServerType;
 const ServerDescription = require('../../../../lib/core/sdam/server_description');
-const ServerSelectors = require('../../../../lib/core/sdam/server_selectors');
+const ServerSelectors = require('../../../../lib/core/sdam/server_selection');
 
 const EJSON = require('mongodb-extjson');