refactor(read-preference): defensively resolve read pref in command

Author: mbroadst

lib/connection/apm.js

@@ -147,6 +147,9 @@ const extractReply = (command, reply) => {
  };
  }
 
+ // in the event of a `noResponse` command, just return
+ if (reply === null) return reply;
+
  return reply.result;
 };
 

lib/connection/pool.js

@@ -1117,8 +1117,6 @@ function _execute(self) {
  var connection = null;
  const connections = self.availableConnections.filter(conn => conn.workItems.length === 0);
 
- // console.log({ totalConnections, connectionsLength: connections.length });
-
  // No connection found that has no work on it, just pick one for pipelining
  if (connections.length === 0) {
  connection =

lib/sdam/server.js

@@ -13,6 +13,35 @@ 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;
+
+// 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;
@@ -67,8 +96,6 @@ class Server extends EventEmitter {
 
  /**
  * Initiate server connect
- *
- * @param {Array} [options.auth] Array of auth options to apply on connect
  */
  connect(options) {
  options = options || {};
@@ -189,7 +216,11 @@ class Server extends EventEmitter {
  // Debug log
  if (this.s.logger.isDebug()) {
  this.s.logger.debug(
- `executing command [${JSON.stringify({ ns, cmd, options })}] against ${this.name}`
+ `executing command [${JSON.stringify({
+ ns,
+ cmd,
+ options: debugOptions(DEBUG_FIELDS, options)
+ })}] against ${this.name}`
  );
  }
 

lib/sdam/topology.js

@@ -137,7 +137,8 @@ class Topology extends EventEmitter {
  // Active client sessions
  sessions: [],
  // Promise library
- promiseLibrary: options.promiseLibrary || Promise
+ promiseLibrary: options.promiseLibrary || Promise,
+ credentials: options.credentialss
  };
 
  // amend options for server instance creation
@@ -198,7 +199,7 @@ class Topology extends EventEmitter {
 
  // otherwise, wait for a server to properly connect based on user provided read preference,
  // or primary.
- const readPreference = options.readPreference || ReadPreference.primary;
+ const readPreference = resolveReadPreference(options);
  this.selectServer(readPreferenceServerSelector(readPreference), (err, server) => {
  if (err) {
  if (typeof callback === 'function') {
@@ -391,6 +392,15 @@ class Topology extends EventEmitter {
  );
  }
 
+ auth(credentials, callback) {
+ if (typeof credentials === 'function') (callback = credentials), (credentials = null);
+ if (typeof callback === 'function') callback(null, true);
+ }
+
+ logout(callback) {
+ if (typeof callback === 'function') callback(null, true);
+ }
+
  // Basic operation support. Eventually this should be moved into command construction
  // during the command refactor.
 
@@ -463,7 +473,10 @@ class Topology extends EventEmitter {
  (callback = options), (options = {}), (options = options || {});
  }
 
- const readPreference = options.readPreference ? options.readPreference : ReadPreference.primary;
+ // TODO: type resolution should happen elsewhere
+ const readPreference = resolveReadPreference(options);
+ options = Object.assign(options, { readPreference });
+
  this.selectServer(readPreferenceServerSelector(readPreference), (err, server) => {
  if (err) {
  callback(err, null);
@@ -686,6 +699,11 @@ function selectServers(topology, selector, timeout, start, callback) {
  // successful iteration, clear the check timer
  clearTimeout(iterationTimer);
 
+ if (topology.description.error) {
+ callback(topology.description.error, null);
+ return;
+ }
+
  // topology description has changed due to monitoring, reattempt server selection
  selectServers(topology, selector, timeout, start, callback);
  };
@@ -779,8 +797,6 @@ function serverErrorEventHandler(server, topology) {
  new monitoring.ServerClosedEvent(topology.s.id, server.description.address)
  );
 
- topology.emit('error', err);
-
  if (err instanceof MongoParseError) {
  resetServerState(server, err, { clearPool: true });
  return;
@@ -843,7 +859,7 @@ function executeWriteOperation(args, options, callback) {
 }
 
 /**
- * Resets the internal state of this server to `Unknown`.
+ * Resets the internal state of this server to `Unknown` by simulating an empty ismaster
  *
  * @private
  * @param {Server} server
@@ -857,7 +873,7 @@ function resetServerState(server, error, options) {
  function resetState() {
  server.emit(
  'descriptionReceived',
- new ServerDescription(server.description.address, null, error)
+ new ServerDescription(server.description.address, null, { error })
  );
  }
 
@@ -869,6 +885,19 @@ function resetServerState(server, error, options) {
  resetState();
 }
 
+function resolveReadPreference(options) {
+ let readPreference = options.readPreference || new ReadPreference('primary');
+ if (typeof readPreference === 'string') {
+ readPreference = new ReadPreference(readPreference);
+ }
+
+ if (!(readPreference instanceof ReadPreference)) {
+ throw new MongoError('read preference must be a ReadPreference instance');
+ }
+
+ return readPreference;
+}
+
 /**
  * A server opening SDAM monitoring event
  *

lib/sdam/topology_description.js

@@ -7,7 +7,6 @@ const ReadPreference = require('../topologies/read_preference');
 const MIN_SUPPORTED_SERVER_VERSION = '2.6';
 const MIN_SUPPORTED_WIRE_VERSION = 2;
 const MAX_SUPPORTED_WIRE_VERSION = 5;
-const DEFAULT_HEARTBEAT_FREQUENCY_MS = 10000;
 
 // An enumeration of topology types we know about
 const TopologyType = {
@@ -29,7 +28,15 @@ class TopologyDescription {
  * @param {number} maxSetVersion
  * @param {ObjectId} maxElectionId
  */
- constructor(topologyType, serverDescriptions, setName, maxSetVersion, maxElectionId, options) {
+ constructor(
+ topologyType,
+ serverDescriptions,
+ setName,
+ maxSetVersion,
+ maxElectionId,
+ options,
+ error
+ ) {
  options = options || {};
 
  // TODO: consider assigning all these values to a temporary value `s` which
@@ -47,6 +54,7 @@ class TopologyDescription {
  this.heartbeatFrequencyMS = options.heartbeatFrequencyMS || 0;
  this.localThresholdMS = options.localThresholdMS || 0;
  this.options = options;
+ this.error = error;
 
  // determine server compatibility
  for (const serverDescription of this.servers.values()) {
@@ -109,6 +117,7 @@ class TopologyDescription {
  let setName = this.setName;
  let maxSetVersion = this.maxSetVersion;
  let maxElectionId = this.maxElectionId;
+ let error = serverDescription.error || null;
 
  const serverType = serverDescription.type;
  let serverDescriptions = new Map(this.servers);
@@ -124,7 +133,8 @@ class TopologyDescription {
  setName,
  maxSetVersion,
  maxElectionId,
- this.options
+ this.options,
+ error
  );
  }
 
@@ -204,7 +214,8 @@ class TopologyDescription {
  setName,
  maxSetVersion,
  maxElectionId,
- this.options
+ this.options,
+ error
  );
  }