om/node_modules/mongodb-core/lib/topologies/mongos.js

'use strict';

const inherits = require('util').inherits;
const f = require('util').format;
const EventEmitter = require('events').EventEmitter;
const BasicCursor = require('../cursor');
const Logger = require('../connection/logger');
const retrieveBSON = require('../connection/utils').retrieveBSON;
const MongoError = require('../error').MongoError;
const Server = require('./server');
const clone = require('./shared').clone;
const diff = require('./shared').diff;
const cloneOptions = require('./shared').cloneOptions;
const createClientInfo = require('./shared').createClientInfo;
const SessionMixins = require('./shared').SessionMixins;
const isRetryableWritesSupported = require('./shared').isRetryableWritesSupported;
const relayEvents = require('../utils').relayEvents;
const isRetryableError = require('../error').isRetryableError;
const BSON = retrieveBSON();

/**
 * @fileOverview The **Mongos** class is a class that represents a Mongos Proxy topology and is
 * used to construct connections.
 *
 * @example
 * var Mongos = require('mongodb-core').Mongos
 *   , ReadPreference = require('mongodb-core').ReadPreference
 *   , assert = require('assert');
 *
 * var server = new Mongos([{host: 'localhost', port: 30000}]);
 * // Wait for the connection event
 * server.on('connect', function(server) {
 *   server.destroy();
 * });
 *
 * // Start connecting
 * server.connect();
 */

//
// States
var DISCONNECTED = 'disconnected';
var CONNECTING = 'connecting';
var CONNECTED = 'connected';
var UNREFERENCED = 'unreferenced';
var DESTROYED = 'destroyed';

function stateTransition(self, newState) {
  var legalTransitions = {
    disconnected: [CONNECTING, DESTROYED, DISCONNECTED],
    connecting: [CONNECTING, DESTROYED, CONNECTED, DISCONNECTED],
    connected: [CONNECTED, DISCONNECTED, DESTROYED, UNREFERENCED],
    unreferenced: [UNREFERENCED, DESTROYED],
    destroyed: [DESTROYED]
  };

  // Get current state
  var legalStates = legalTransitions[self.state];
  if (legalStates && legalStates.indexOf(newState) !== -1) {
    self.state = newState;
  } else {
    self.logger.error(
      f(
        'Pool with id [%s] failed attempted illegal state transition from [%s] to [%s] only following state allowed [%s]',
        self.id,
        self.state,
        newState,
        legalStates
      )
    );
  }
}

//
// ReplSet instance id
var id = 1;
var handlers = ['connect', 'close', 'error', 'timeout', 'parseError'];

/**
 * Creates a new Mongos instance
 * @class
 * @param {array} seedlist A list of seeds for the replicaset
 * @param {number} [options.haInterval=5000] The High availability period for replicaset inquiry
 * @param {Cursor} [options.cursorFactory=Cursor] The cursor factory class used for all query cursors
 * @param {number} [options.size=5] Server connection pool size
 * @param {boolean} [options.keepAlive=true] TCP Connection keep alive enabled
 * @param {number} [options.keepAliveInitialDelay=0] Initial delay before TCP keep alive enabled
 * @param {number} [options.localThresholdMS=15] Cutoff latency point in MS for MongoS proxy selection
 * @param {boolean} [options.noDelay=true] TCP Connection no delay
 * @param {number} [options.connectionTimeout=1000] TCP Connection timeout setting
 * @param {number} [options.socketTimeout=0] TCP Socket timeout setting
 * @param {boolean} [options.ssl=false] Use SSL for connection
 * @param {boolean|function} [options.checkServerIdentity=true] Ensure we check server identify during SSL, set to false to disable checking. Only works for Node 0.12.x or higher. You can pass in a boolean or your own checkServerIdentity override function.
 * @param {Buffer} [options.ca] SSL Certificate store binary buffer
 * @param {Buffer} [options.crl] SSL Certificate revocation store binary buffer
 * @param {Buffer} [options.cert] SSL Certificate binary buffer
 * @param {Buffer} [options.key] SSL Key file binary buffer
 * @param {string} [options.passphrase] SSL Certificate pass phrase
 * @param {string} [options.servername=null] String containing the server name requested via TLS SNI.
 * @param {boolean} [options.rejectUnauthorized=true] Reject unauthorized server certificates
 * @param {boolean} [options.promoteLongs=true] Convert Long values from the db into Numbers if they fit into 53 bits
 * @param {boolean} [options.promoteValues=true] Promotes BSON values to native types where possible, set to false to only receive wrapper types.
 * @param {boolean} [options.promoteBuffers=false] Promotes Binary BSON values to native Node Buffers.
 * @param {boolean} [options.domainsEnabled=false] Enable the wrapping of the callback in the current domain, disabled by default to avoid perf hit.
 * @param {boolean} [options.monitorCommands=false] Enable command monitoring for this topology
 * @return {Mongos} A cursor instance
 * @fires Mongos#connect
 * @fires Mongos#reconnect
 * @fires Mongos#joined
 * @fires Mongos#left
 * @fires Mongos#failed
 * @fires Mongos#fullsetup
 * @fires Mongos#all
 * @fires Mongos#serverHeartbeatStarted
 * @fires Mongos#serverHeartbeatSucceeded
 * @fires Mongos#serverHeartbeatFailed
 * @fires Mongos#topologyOpening
 * @fires Mongos#topologyClosed
 * @fires Mongos#topologyDescriptionChanged
 * @property {string} type the topology type.
 * @property {string} parserType the parser type used (c++ or js).
 */
var Mongos = function(seedlist, options) {
  options = options || {};

  // Get replSet Id
  this.id = id++;

  // Internal state
  this.s = {
    options: Object.assign({}, options),
    // BSON instance
    bson:
      options.bson ||
      new BSON([
        BSON.Binary,
        BSON.Code,
        BSON.DBRef,
        BSON.Decimal128,
        BSON.Double,
        BSON.Int32,
        BSON.Long,
        BSON.Map,
        BSON.MaxKey,
        BSON.MinKey,
        BSON.ObjectId,
        BSON.BSONRegExp,
        BSON.Symbol,
        BSON.Timestamp
      ]),
    // Factory overrides
    Cursor: options.cursorFactory || BasicCursor,
    // Logger instance
    logger: Logger('Mongos', options),
    // Seedlist
    seedlist: seedlist,
    // Ha interval
    haInterval: options.haInterval ? options.haInterval : 10000,
    // Disconnect handler
    disconnectHandler: options.disconnectHandler,
    // Server selection index
    index: 0,
    // Connect function options passed in
    connectOptions: {},
    // Are we running in debug mode
    debug: typeof options.debug === 'boolean' ? options.debug : false,
    // localThresholdMS
    localThresholdMS: options.localThresholdMS || 15,
    // Client info
    clientInfo: createClientInfo(options)
  };

  // Set the client info
  this.s.options.clientInfo = createClientInfo(options);

  // Log info warning if the socketTimeout < haInterval as it will cause
  // a lot of recycled connections to happen.
  if (
    this.s.logger.isWarn() &&
    this.s.options.socketTimeout !== 0 &&
    this.s.options.socketTimeout < this.s.haInterval
  ) {
    this.s.logger.warn(
      f(
        'warning socketTimeout %s is less than haInterval %s. This might cause unnecessary server reconnections due to socket timeouts',
        this.s.options.socketTimeout,
        this.s.haInterval
      )
    );
  }

  // Disconnected state
  this.state = DISCONNECTED;

  // Current proxies we are connecting to
  this.connectingProxies = [];
  // Currently connected proxies
  this.connectedProxies = [];
  // Disconnected proxies
  this.disconnectedProxies = [];
  // Index of proxy to run operations against
  this.index = 0;
  // High availability timeout id
  this.haTimeoutId = null;
  // Last ismaster
  this.ismaster = null;

  // Description of the Replicaset
  this.topologyDescription = {
    topologyType: 'Unknown',
    servers: []
  };

  // Highest clusterTime seen in responses from the current deployment
  this.clusterTime = null;

  // Add event listener
  EventEmitter.call(this);
};

inherits(Mongos, EventEmitter);
Object.assign(Mongos.prototype, SessionMixins);

Object.defineProperty(Mongos.prototype, 'type', {
  enumerable: true,
  get: function() {
    return 'mongos';
  }
});

Object.defineProperty(Mongos.prototype, 'parserType', {
  enumerable: true,
  get: function() {
    return BSON.native ? 'c++' : 'js';
  }
});

Object.defineProperty(Mongos.prototype, 'logicalSessionTimeoutMinutes', {
  enumerable: true,
  get: function() {
    if (!this.ismaster) return null;
    return this.ismaster.logicalSessionTimeoutMinutes || null;
  }
});

/**
 * Emit event if it exists
 * @method
 */
function emitSDAMEvent(self, event, description) {
  if (self.listeners(event).length > 0) {
    self.emit(event, description);
  }
}

const SERVER_EVENTS = ['serverDescriptionChanged', 'error', 'close', 'timeout', 'parseError'];
function destroyServer(server, options, callback) {
  options = options || {};
  SERVER_EVENTS.forEach(event => server.removeAllListeners(event));
  server.destroy(options, callback);
}

/**
 * Initiate server connect
 */
Mongos.prototype.connect = function(options) {
  var self = this;
  // Add any connect level options to the internal state
  this.s.connectOptions = options || {};

  // Set connecting state
  stateTransition(this, CONNECTING);

  // Create server instances
  var servers = this.s.seedlist.map(function(x) {
    const server = new Server(
      Object.assign({}, self.s.options, x, options, {
        reconnect: false,
        monitoring: false,
        parent: self,
        clientInfo: clone(self.s.clientInfo)
      })
    );

    relayEvents(server, self, ['serverDescriptionChanged']);
    return server;
  });

  // Emit the topology opening event
  emitSDAMEvent(this, 'topologyOpening', { topologyId: this.id });

  // Start all server connections
  connectProxies(self, servers);
};

/**
 * Authenticate the topology.
 * @method
 * @param {MongoCredentials} credentials The credentials for authentication we are using
 * @param {authResultCallback} callback A callback function
 */
Mongos.prototype.auth = function(credentials, callback) {
  if (typeof callback === 'function') callback(null, null);
};

function handleEvent(self) {
  return function() {
    if (self.state === DESTROYED) return;
    // Move to list of disconnectedProxies
    moveServerFrom(self.connectedProxies, self.disconnectedProxies, this);
    // Emit the initial topology
    emitTopologyDescriptionChanged(self);
    // Emit the left signal
    self.emit('left', 'mongos', this);
    // Emit the sdam event
    self.emit('serverClosed', {
      topologyId: self.id,
      address: this.name
    });
  };
}

function handleInitialConnectEvent(self, event) {
  return function() {
    var _this = this;

    // Destroy the instance
    if (self.state === DESTROYED) {
      // Emit the initial topology
      emitTopologyDescriptionChanged(self);
      // Move from connectingProxies
      moveServerFrom(self.connectingProxies, self.disconnectedProxies, this);
      return this.destroy();
    }

    // Check the type of server
    if (event === 'connect') {
      // Get last known ismaster
      self.ismaster = _this.lastIsMaster();

      // Is this not a proxy, remove t
      if (self.ismaster.msg === 'isdbgrid') {
        // Add to the connectd list
        for (let i = 0; i < self.connectedProxies.length; i++) {
          if (self.connectedProxies[i].name === _this.name) {
            // Move from connectingProxies
            moveServerFrom(self.connectingProxies, self.disconnectedProxies, _this);
            // Emit the initial topology
            emitTopologyDescriptionChanged(self);
            _this.destroy();
            return self.emit('failed', _this);
          }
        }

        // Remove the handlers
        for (let i = 0; i < handlers.length; i++) {
          _this.removeAllListeners(handlers[i]);
        }

        // Add stable state handlers
        _this.on('error', handleEvent(self, 'error'));
        _this.on('close', handleEvent(self, 'close'));
        _this.on('timeout', handleEvent(self, 'timeout'));
        _this.on('parseError', handleEvent(self, 'parseError'));

        // Move from connecting proxies connected
        moveServerFrom(self.connectingProxies, self.connectedProxies, _this);
        // Emit the joined event
        self.emit('joined', 'mongos', _this);
      } else {
        // Print warning if we did not find a mongos proxy
        if (self.s.logger.isWarn()) {
          var message = 'expected mongos proxy, but found replicaset member mongod for server %s';
          // We have a standalone server
          if (!self.ismaster.hosts) {
            message = 'expected mongos proxy, but found standalone mongod for server %s';
          }

          self.s.logger.warn(f(message, _this.name));
        }

        // This is not a mongos proxy, remove it completely
        removeProxyFrom(self.connectingProxies, _this);
        // Emit the left event
        self.emit('left', 'server', _this);
        // Emit failed event
        self.emit('failed', _this);
      }
    } else {
      moveServerFrom(self.connectingProxies, self.disconnectedProxies, this);
      // Emit the left event
      self.emit('left', 'mongos', this);
      // Emit failed event
      self.emit('failed', this);
    }

    // Emit the initial topology
    emitTopologyDescriptionChanged(self);

    // Trigger topologyMonitor
    if (self.connectingProxies.length === 0) {
      // Emit connected if we are connected
      if (self.connectedProxies.length > 0 && self.state === CONNECTING) {
        // Set the state to connected
        stateTransition(self, CONNECTED);
        // Emit the connect event
        self.emit('connect', self);
        self.emit('fullsetup', self);
        self.emit('all', self);
      } else if (self.disconnectedProxies.length === 0) {
        // Print warning if we did not find a mongos proxy
        if (self.s.logger.isWarn()) {
          self.s.logger.warn(
            f('no mongos proxies found in seed list, did you mean to connect to a replicaset')
          );
        }

        // Emit the error that no proxies were found
        return self.emit('error', new MongoError('no mongos proxies found in seed list'));
      }

      // Topology monitor
      topologyMonitor(self, { firstConnect: true });
    }
  };
}

function connectProxies(self, servers) {
  // Update connectingProxies
  self.connectingProxies = self.connectingProxies.concat(servers);

  // Index used to interleaf the server connects, avoiding
  // runtime issues on io constrained vm's
  var timeoutInterval = 0;

  function connect(server, timeoutInterval) {
    setTimeout(function() {
      // Emit opening server event
      self.emit('serverOpening', {
        topologyId: self.id,
        address: server.name
      });

      // Emit the initial topology
      emitTopologyDescriptionChanged(self);

      // Add event handlers
      server.once('close', handleInitialConnectEvent(self, 'close'));
      server.once('timeout', handleInitialConnectEvent(self, 'timeout'));
      server.once('parseError', handleInitialConnectEvent(self, 'parseError'));
      server.once('error', handleInitialConnectEvent(self, 'error'));
      server.once('connect', handleInitialConnectEvent(self, 'connect'));

      // Command Monitoring events
      relayEvents(server, self, ['commandStarted', 'commandSucceeded', 'commandFailed']);

      // Start connection
      server.connect(self.s.connectOptions);
    }, timeoutInterval);
  }
  // Start all the servers
  while (servers.length > 0) {
    connect(servers.shift(), timeoutInterval++);
  }
}

function pickProxy(self, session) {
  // TODO: Destructure :)
  const transaction = session && session.transaction;

  if (transaction && transaction.server) {
    if (transaction.server.isConnected()) {
      return transaction.server;
    } else {
      transaction.unpinServer();
    }
  }

  // Get the currently connected Proxies
  var connectedProxies = self.connectedProxies.slice(0);

  // Set lower bound
  var lowerBoundLatency = Number.MAX_VALUE;

  // Determine the lower bound for the Proxies
  for (var i = 0; i < connectedProxies.length; i++) {
    if (connectedProxies[i].lastIsMasterMS < lowerBoundLatency) {
      lowerBoundLatency = connectedProxies[i].lastIsMasterMS;
    }
  }

  // Filter out the possible servers
  connectedProxies = connectedProxies.filter(function(server) {
    if (
      server.lastIsMasterMS <= lowerBoundLatency + self.s.localThresholdMS &&
      server.isConnected()
    ) {
      return true;
    }
  });

  let proxy;

  // We have no connectedProxies pick first of the connected ones
  if (connectedProxies.length === 0) {
    proxy = self.connectedProxies[0];
  } else {
    // Get proxy
    proxy = connectedProxies[self.index % connectedProxies.length];
    // Update the index
    self.index = (self.index + 1) % connectedProxies.length;
  }

  if (transaction && transaction.isActive && proxy && proxy.isConnected()) {
    transaction.pinServer(proxy);
  }

  // Return the proxy
  return proxy;
}

function moveServerFrom(from, to, proxy) {
  for (var i = 0; i < from.length; i++) {
    if (from[i].name === proxy.name) {
      from.splice(i, 1);
    }
  }

  for (i = 0; i < to.length; i++) {
    if (to[i].name === proxy.name) {
      to.splice(i, 1);
    }
  }

  to.push(proxy);
}

function removeProxyFrom(from, proxy) {
  for (var i = 0; i < from.length; i++) {
    if (from[i].name === proxy.name) {
      from.splice(i, 1);
    }
  }
}

function reconnectProxies(self, proxies, callback) {
  // Count lefts
  var count = proxies.length;

  // Handle events
  var _handleEvent = function(self, event) {
    return function() {
      var _self = this;
      count = count - 1;

      // Destroyed
      if (self.state === DESTROYED || self.state === UNREFERENCED) {
        moveServerFrom(self.connectingProxies, self.disconnectedProxies, _self);
        return this.destroy();
      }

      if (event === 'connect') {
        // Destroyed
        if (self.state === DESTROYED || self.state === UNREFERENCED) {
          moveServerFrom(self.connectingProxies, self.disconnectedProxies, _self);
          return _self.destroy();
        }

        // Remove the handlers
        for (var i = 0; i < handlers.length; i++) {
          _self.removeAllListeners(handlers[i]);
        }

        // Add stable state handlers
        _self.on('error', handleEvent(self, 'error'));
        _self.on('close', handleEvent(self, 'close'));
        _self.on('timeout', handleEvent(self, 'timeout'));
        _self.on('parseError', handleEvent(self, 'parseError'));

        // Move to the connected servers
        moveServerFrom(self.connectingProxies, self.connectedProxies, _self);
        // Emit topology Change
        emitTopologyDescriptionChanged(self);
        // Emit joined event
        self.emit('joined', 'mongos', _self);
      } else {
        // Move from connectingProxies
        moveServerFrom(self.connectingProxies, self.disconnectedProxies, _self);
        this.destroy();
      }

      // Are we done finish up callback
      if (count === 0) {
        callback();
      }
    };
  };

  // No new servers
  if (count === 0) {
    return callback();
  }

  // Execute method
  function execute(_server, i) {
    setTimeout(function() {
      // Destroyed
      if (self.state === DESTROYED || self.state === UNREFERENCED) {
        return;
      }

      // Create a new server instance
      var server = new Server(
        Object.assign({}, self.s.options, {
          host: _server.name.split(':')[0],
          port: parseInt(_server.name.split(':')[1], 10),
          reconnect: false,
          monitoring: false,
          parent: self,
          clientInfo: clone(self.s.clientInfo)
        })
      );

      destroyServer(_server);
      removeProxyFrom(self.disconnectedProxies, _server);

      // Relay the server description change
      relayEvents(server, self, ['serverDescriptionChanged']);

      // Emit opening server event
      self.emit('serverOpening', {
        topologyId: server.s.topologyId !== -1 ? server.s.topologyId : self.id,
        address: server.name
      });

      // Add temp handlers
      server.once('connect', _handleEvent(self, 'connect'));
      server.once('close', _handleEvent(self, 'close'));
      server.once('timeout', _handleEvent(self, 'timeout'));
      server.once('error', _handleEvent(self, 'error'));
      server.once('parseError', _handleEvent(self, 'parseError'));

      // Command Monitoring events
      relayEvents(server, self, ['commandStarted', 'commandSucceeded', 'commandFailed']);

      // Connect to proxy
      self.connectingProxies.push(server);
      server.connect(self.s.connectOptions);
    }, i);
  }

  // Create new instances
  for (var i = 0; i < proxies.length; i++) {
    execute(proxies[i], i);
  }
}

function topologyMonitor(self, options) {
  options = options || {};

  // Set momitoring timeout
  self.haTimeoutId = setTimeout(function() {
    if (self.state === DESTROYED || self.state === UNREFERENCED) return;
    // If we have a primary and a disconnect handler, execute
    // buffered operations
    if (self.isConnected() && self.s.disconnectHandler) {
      self.s.disconnectHandler.execute();
    }

    // Get the connectingServers
    var proxies = self.connectedProxies.slice(0);
    // Get the count
    var count = proxies.length;

    // If the count is zero schedule a new fast
    function pingServer(_self, _server, cb) {
      // Measure running time
      var start = new Date().getTime();

      // Emit the server heartbeat start
      emitSDAMEvent(self, 'serverHeartbeatStarted', { connectionId: _server.name });

      // Execute ismaster
      _server.command(
        'admin.$cmd',
        {
          ismaster: true
        },
        {
          monitoring: true,
          socketTimeout: self.s.options.connectionTimeout || 2000
        },
        function(err, r) {
          if (self.state === DESTROYED || self.state === UNREFERENCED) {
            // Move from connectingProxies
            moveServerFrom(self.connectedProxies, self.disconnectedProxies, _server);
            _server.destroy();
            return cb(err, r);
          }

          // Calculate latency
          var latencyMS = new Date().getTime() - start;

          // We had an error, remove it from the state
          if (err) {
            // Emit the server heartbeat failure
            emitSDAMEvent(self, 'serverHeartbeatFailed', {
              durationMS: latencyMS,
              failure: err,
              connectionId: _server.name
            });
            // Move from connected proxies to disconnected proxies
            moveServerFrom(self.connectedProxies, self.disconnectedProxies, _server);
          } else {
            // Update the server ismaster
            _server.ismaster = r.result;
            _server.lastIsMasterMS = latencyMS;

            // Server heart beat event
            emitSDAMEvent(self, 'serverHeartbeatSucceeded', {
              durationMS: latencyMS,
              reply: r.result,
              connectionId: _server.name
            });
          }

          cb(err, r);
        }
      );
    }

    // No proxies initiate monitor again
    if (proxies.length === 0) {
      // Emit close event if any listeners registered
      if (self.listeners('close').length > 0 && self.state === CONNECTING) {
        self.emit('error', new MongoError('no mongos proxy available'));
      } else {
        self.emit('close', self);
      }

      // Attempt to connect to any unknown servers
      return reconnectProxies(self, self.disconnectedProxies, function() {
        if (self.state === DESTROYED || self.state === UNREFERENCED) return;

        // Are we connected ? emit connect event
        if (self.state === CONNECTING && options.firstConnect) {
          self.emit('connect', self);
          self.emit('fullsetup', self);
          self.emit('all', self);
        } else if (self.isConnected()) {
          self.emit('reconnect', self);
        } else if (!self.isConnected() && self.listeners('close').length > 0) {
          self.emit('close', self);
        }

        // Perform topology monitor
        topologyMonitor(self);
      });
    }

    // Ping all servers
    for (var i = 0; i < proxies.length; i++) {
      pingServer(self, proxies[i], function() {
        count = count - 1;

        if (count === 0) {
          if (self.state === DESTROYED || self.state === UNREFERENCED) return;

          // Attempt to connect to any unknown servers
          reconnectProxies(self, self.disconnectedProxies, function() {
            if (self.state === DESTROYED || self.state === UNREFERENCED) return;
            // Perform topology monitor
            topologyMonitor(self);
          });
        }
      });
    }
  }, self.s.haInterval);
}

/**
 * Returns the last known ismaster document for this server
 * @method
 * @return {object}
 */
Mongos.prototype.lastIsMaster = function() {
  return this.ismaster;
};

/**
 * Unref all connections belong to this server
 * @method
 */
Mongos.prototype.unref = function() {
  // Transition state
  stateTransition(this, UNREFERENCED);
  // Get all proxies
  var proxies = this.connectedProxies.concat(this.connectingProxies);
  proxies.forEach(function(x) {
    x.unref();
  });

  clearTimeout(this.haTimeoutId);
};

/**
 * Destroy the server connection
 * @param {boolean} [options.force=false] Force destroy the pool
 * @method
 */
Mongos.prototype.destroy = function(options, callback) {
  if (this.haTimeoutId) {
    clearTimeout(this.haTimeoutId);
  }

  const proxies = this.connectedProxies.concat(this.connectingProxies);
  let serverCount = proxies.length;
  const serverDestroyed = () => {
    serverCount--;
    if (serverCount > 0) {
      return;
    }

    emitTopologyDescriptionChanged(this);
    emitSDAMEvent(this, 'topologyClosed', { topologyId: this.id });
    stateTransition(this, DESTROYED);
    if (typeof callback === 'function') {
      callback(null, null);
    }
  };

  if (serverCount === 0) {
    serverDestroyed();
    return;
  }

  // Destroy all connecting servers
  proxies.forEach(server => {
    // Emit the sdam event
    this.emit('serverClosed', {
      topologyId: this.id,
      address: server.name
    });

    destroyServer(server, options, serverDestroyed);
    moveServerFrom(this.connectedProxies, this.disconnectedProxies, server);
  });
};

/**
 * Figure out if the server is connected
 * @method
 * @return {boolean}
 */
Mongos.prototype.isConnected = function() {
  return this.connectedProxies.length > 0;
};

/**
 * Figure out if the server instance was destroyed by calling destroy
 * @method
 * @return {boolean}
 */
Mongos.prototype.isDestroyed = function() {
  return this.state === DESTROYED;
};

//
// Operations
//

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 self = args.self;
  const op = args.op;
  const ns = args.ns;
  const ops = args.ops;

  // Pick a server
  let server = pickProxy(self, options.session);
  // No server found error out
  if (!server) return callback(new MongoError('no mongos proxy available'));

  const willRetryWrite =
    !args.retrying &&
    !!options.retryWrites &&
    options.session &&
    isRetryableWritesSupported(self) &&
    !options.session.inTransaction();

  const handler = (err, result) => {
    if (!err) return callback(null, result);
    if (!isRetryableError(err) || !willRetryWrite) {
      return callback(err);
    }

    // Pick another server
    server = pickProxy(self, options.session);

    // No server found error out with original error
    if (!server) {
      return callback(err);
    }

    const newArgs = Object.assign({}, args, { retrying: true });
    return executeWriteOperation(newArgs, options, callback);
  };

  if (callback.operationId) {
    handler.operationId = callback.operationId;
  }

  // increment and assign txnNumber
  if (willRetryWrite) {
    options.session.incrementTransactionNumber();
    options.willRetryWrite = willRetryWrite;
  }

  // rerun the operation
  server[op](ns, ops, options, handler);
}

/**
 * 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 {boolean} [options.retryWrites] Enable retryable writes for this operation
 * @param {opResultCallback} callback A callback function
 */
Mongos.prototype.insert = function(ns, ops, options, callback) {
  if (typeof options === 'function') {
    (callback = options), (options = {}), (options = options || {});
  }

  if (this.state === DESTROYED) return callback(new MongoError(f('topology was destroyed')));

  // Not connected but we have a disconnecthandler
  if (!this.isConnected() && this.s.disconnectHandler != null) {
    return this.s.disconnectHandler.add('insert', ns, ops, options, callback);
  }

  // No mongos proxy available
  if (!this.isConnected()) {
    return callback(new MongoError('no mongos proxy available'));
  }

  // Execute write operation
  executeWriteOperation({ self: 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 {boolean} [options.retryWrites] Enable retryable writes for this operation
 * @param {opResultCallback} callback A callback function
 */
Mongos.prototype.update = function(ns, ops, options, callback) {
  if (typeof options === 'function') {
    (callback = options), (options = {}), (options = options || {});
  }

  if (this.state === DESTROYED) return callback(new MongoError(f('topology was destroyed')));

  // Not connected but we have a disconnecthandler
  if (!this.isConnected() && this.s.disconnectHandler != null) {
    return this.s.disconnectHandler.add('update', ns, ops, options, callback);
  }

  // No mongos proxy available
  if (!this.isConnected()) {
    return callback(new MongoError('no mongos proxy available'));
  }

  // Execute write operation
  executeWriteOperation({ self: 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 {boolean} [options.retryWrites] Enable retryable writes for this operation
 * @param {opResultCallback} callback A callback function
 */
Mongos.prototype.remove = function(ns, ops, options, callback) {
  if (typeof options === 'function') {
    (callback = options), (options = {}), (options = options || {});
  }

  if (this.state === DESTROYED) return callback(new MongoError(f('topology was destroyed')));

  // Not connected but we have a disconnecthandler
  if (!this.isConnected() && this.s.disconnectHandler != null) {
    return this.s.disconnectHandler.add('remove', ns, ops, options, callback);
  }

  // No mongos proxy available
  if (!this.isConnected()) {
    return callback(new MongoError('no mongos proxy available'));
  }

  // Execute write operation
  executeWriteOperation({ self: this, op: 'remove', ns, ops }, options, callback);
};

const RETRYABLE_WRITE_OPERATIONS = ['findAndModify', 'insert', 'update', 'delete'];

function isWriteCommand(command) {
  return RETRYABLE_WRITE_OPERATIONS.some(op => command[op]);
}

/**
 * Execute a command
 * @method
 * @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 {Connection} [options.connection] Specify connection object to execute command against
 * @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
 */
Mongos.prototype.command = function(ns, cmd, options, callback) {
  if (typeof options === 'function') {
    (callback = options), (options = {}), (options = options || {});
  }

  if (this.state === DESTROYED) return callback(new MongoError(f('topology was destroyed')));
  var self = this;

  // Pick a proxy
  var server = pickProxy(self, options.session);

  // Topology is not connected, save the call in the provided store to be
  // Executed at some point when the handler deems it's reconnected
  if ((server == null || !server.isConnected()) && this.s.disconnectHandler != null) {
    return this.s.disconnectHandler.add('command', ns, cmd, options, callback);
  }

  // No server returned we had an error
  if (server == null) {
    return callback(new MongoError('no mongos proxy available'));
  }

  // Cloned options
  var clonedOptions = cloneOptions(options);
  clonedOptions.topology = self;

  const willRetryWrite =
    !options.retrying &&
    options.retryWrites &&
    options.session &&
    isRetryableWritesSupported(self) &&
    !options.session.inTransaction() &&
    isWriteCommand(cmd);

  const cb = (err, result) => {
    if (!err) return callback(null, result);
    if (!isRetryableError(err)) {
      return callback(err);
    }

    if (willRetryWrite) {
      const newOptions = Object.assign({}, clonedOptions, { retrying: true });
      return this.command(ns, cmd, newOptions, callback);
    }

    return callback(err);
  };

  // increment and assign txnNumber
  if (willRetryWrite) {
    options.session.incrementTransactionNumber();
    options.willRetryWrite = willRetryWrite;
  }

  // Execute the command
  server.command(ns, cmd, clonedOptions, cb);
};

/**
 * Get a new cursor
 * @method
 * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
 * @param {object|Long} cmd Can be either a command returning a cursor or a cursorId
 * @param {object} [options] Options for the cursor
 * @param {object} [options.batchSize=0] Batchsize for the operation
 * @param {array} [options.documents=[]] Initial documents list for cursor
 * @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.ignoreUndefined=false] Specify if the BSON serializer should ignore undefined fields.
 * @param {ClientSession} [options.session=null] Session to use for the operation
 * @param {object} [options.topology] The internal topology of the created cursor
 * @returns {Cursor}
 */
Mongos.prototype.cursor = function(ns, cmd, options) {
  options = options || {};
  const topology = options.topology || this;

  // Set up final cursor type
  var FinalCursor = options.cursorFactory || this.s.Cursor;

  // Return the cursor
  return new FinalCursor(this.s.bson, ns, cmd, options, topology, this.s.options);
};

/**
 * Selects a server
 *
 * @method
 * @param {function} selector Unused
 * @param {ReadPreference} [options.readPreference] Unused
 * @param {ClientSession} [options.session] Specify a session if it is being used
 * @param {function} callback
 */
Mongos.prototype.selectServer = function(selector, options, callback) {
  if (typeof selector === 'function' && typeof callback === 'undefined')
    (callback = selector), (selector = undefined), (options = {});
  if (typeof options === 'function')
    (callback = options), (options = selector), (selector = undefined);
  options = options || {};

  const server = pickProxy(this, options.session);
  if (this.s.debug) this.emit('pickedServer', null, server);
  callback(null, server);
};

/**
 * All raw connections
 * @method
 * @return {Connection[]}
 */
Mongos.prototype.connections = function() {
  var connections = [];

  for (var i = 0; i < this.connectedProxies.length; i++) {
    connections = connections.concat(this.connectedProxies[i].connections());
  }

  return connections;
};

function emitTopologyDescriptionChanged(self) {
  if (self.listeners('topologyDescriptionChanged').length > 0) {
    var topology = 'Unknown';
    if (self.connectedProxies.length > 0) {
      topology = 'Sharded';
    }

    // Generate description
    var description = {
      topologyType: topology,
      servers: []
    };

    // All proxies
    var proxies = self.disconnectedProxies.concat(self.connectingProxies);

    // Add all the disconnected proxies
    description.servers = description.servers.concat(
      proxies.map(function(x) {
        var description = x.getDescription();
        description.type = 'Unknown';
        return description;
      })
    );

    // Add all the connected proxies
    description.servers = description.servers.concat(
      self.connectedProxies.map(function(x) {
        var description = x.getDescription();
        description.type = 'Mongos';
        return description;
      })
    );

    // Get the diff
    var diffResult = diff(self.topologyDescription, description);

    // Create the result
    var result = {
      topologyId: self.id,
      previousDescription: self.topologyDescription,
      newDescription: description,
      diff: diffResult
    };

    // Emit the topologyDescription change
    if (diffResult.servers.length > 0) {
      self.emit('topologyDescriptionChanged', result);
    }

    // Set the new description
    self.topologyDescription = description;
  }
}

/**
 * A mongos connect event, used to verify that the connection is up and running
 *
 * @event Mongos#connect
 * @type {Mongos}
 */

/**
 * A mongos reconnect event, used to verify that the mongos topology has reconnected
 *
 * @event Mongos#reconnect
 * @type {Mongos}
 */

/**
 * A mongos fullsetup event, used to signal that all topology members have been contacted.
 *
 * @event Mongos#fullsetup
 * @type {Mongos}
 */

/**
 * A mongos all event, used to signal that all topology members have been contacted.
 *
 * @event Mongos#all
 * @type {Mongos}
 */

/**
 * A server member left the mongos list
 *
 * @event Mongos#left
 * @type {Mongos}
 * @param {string} type The type of member that left (mongos)
 * @param {Server} server The server object that left
 */

/**
 * A server member joined the mongos list
 *
 * @event Mongos#joined
 * @type {Mongos}
 * @param {string} type The type of member that left (mongos)
 * @param {Server} server The server object that joined
 */

/**
 * A server opening SDAM monitoring event
 *
 * @event Mongos#serverOpening
 * @type {object}
 */

/**
 * A server closed SDAM monitoring event
 *
 * @event Mongos#serverClosed
 * @type {object}
 */

/**
 * A server description SDAM change monitoring event
 *
 * @event Mongos#serverDescriptionChanged
 * @type {object}
 */

/**
 * A topology open SDAM event
 *
 * @event Mongos#topologyOpening
 * @type {object}
 */

/**
 * A topology closed SDAM event
 *
 * @event Mongos#topologyClosed
 * @type {object}
 */

/**
 * A topology structure SDAM change event
 *
 * @event Mongos#topologyDescriptionChanged
 * @type {object}
 */

/**
 * A topology serverHeartbeatStarted SDAM event
 *
 * @event Mongos#serverHeartbeatStarted
 * @type {object}
 */

/**
 * A topology serverHeartbeatFailed SDAM event
 *
 * @event Mongos#serverHeartbeatFailed
 * @type {object}
 */

/**
 * A topology serverHeartbeatSucceeded SDAM change event
 *
 * @event Mongos#serverHeartbeatSucceeded
 * @type {object}
 */

/**
 * An event emitted indicating a command was started, if command monitoring is enabled
 *
 * @event Mongos#commandStarted
 * @type {object}
 */

/**
 * An event emitted indicating a command succeeded, if command monitoring is enabled
 *
 * @event Mongos#commandSucceeded
 * @type {object}
 */

/**
 * An event emitted indicating a command failed, if command monitoring is enabled
 *
 * @event Mongos#commandFailed
 * @type {object}
 */

module.exports = Mongos;