'use strict'; const Server = require('./server'); const Cursor = require('../cursor'); const MongoError = require('../core').MongoError; const TopologyBase = require('./topology_base').TopologyBase; const Store = require('./topology_base').Store; const CReplSet = require('../core').ReplSet; const MAX_JS_INT = require('../utils').MAX_JS_INT; const translateOptions = require('../utils').translateOptions; const filterOptions = require('../utils').filterOptions; const mergeOptions = require('../utils').mergeOptions; /** * @fileOverview The **ReplSet** class is a class that represents a Replicaset topology and is * used to construct connections. * * **ReplSet Should not be used, use MongoClient.connect** */ // Allowed parameters var legalOptionNames = [ 'ha', 'haInterval', 'replicaSet', 'rs_name', 'secondaryAcceptableLatencyMS', 'connectWithNoPrimary', 'poolSize', 'ssl', 'checkServerIdentity', 'sslValidate', 'sslCA', 'sslCert', 'ciphers', 'ecdhCurve', 'sslCRL', 'sslKey', 'sslPass', 'socketOptions', 'bufferMaxEntries', 'store', 'auto_reconnect', 'autoReconnect', 'emitError', 'keepAlive', 'keepAliveInitialDelay', 'noDelay', 'connectTimeoutMS', 'socketTimeoutMS', 'strategy', 'debug', 'family', 'loggerLevel', 'logger', 'reconnectTries', 'appname', 'domainsEnabled', 'servername', 'promoteLongs', 'promoteValues', 'promoteBuffers', 'bsonRegExp', 'maxStalenessSeconds', 'promiseLibrary', 'minSize', 'monitorCommands' ]; /** * Creates a new ReplSet instance * @class * @deprecated * @param {Server[]} servers A seedlist of servers participating in the replicaset. * @param {object} [options] Optional settings. * @param {boolean} [options.ha=true] Turn on high availability monitoring. * @param {number} [options.haInterval=10000] Time between each replicaset status check. * @param {string} [options.replicaSet] The name of the replicaset to connect to. * @param {number} [options.secondaryAcceptableLatencyMS=15] Sets the range of servers to pick when using NEAREST (lowest ping ms + the latency fence, ex: range of 1 to (1 + 15) ms) * @param {boolean} [options.connectWithNoPrimary=false] Sets if the driver should connect even if no primary is available * @param {number} [options.poolSize=5] Number of connections in the connection pool for each server instance, set to 5 as default for legacy reasons. * @param {boolean} [options.ssl=false] Use ssl connection (needs to have a mongod server with ssl support) * @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 {boolean} [options.sslValidate=false] Validate mongod server certificate against ca (needs to have a mongod server with ssl support, 2.4 or higher) * @param {array} [options.sslCA] Array of valid certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher) * @param {array} [options.sslCRL] Array of revocation certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher) * @param {(Buffer|string)} [options.sslCert] String or buffer containing the certificate we wish to present (needs to have a mongod server with ssl support, 2.4 or higher. * @param {string} [options.ciphers] Passed directly through to tls.createSecureContext. See https://nodejs.org/dist/latest-v9.x/docs/api/tls.html#tls_tls_createsecurecontext_options for more info. * @param {string} [options.ecdhCurve] Passed directly through to tls.createSecureContext. See https://nodejs.org/dist/latest-v9.x/docs/api/tls.html#tls_tls_createsecurecontext_options for more info. * @param {(Buffer|string)} [options.sslKey] String or buffer containing the certificate private key we wish to present (needs to have a mongod server with ssl support, 2.4 or higher) * @param {(Buffer|string)} [options.sslPass] String or buffer containing the certificate password (needs to have a mongod server with ssl support, 2.4 or higher) * @param {string} [options.servername] String containing the server name requested via TLS SNI. * @param {object} [options.socketOptions] Socket options * @param {boolean} [options.socketOptions.noDelay=true] TCP Socket NoDelay option. * @param {boolean} [options.socketOptions.keepAlive=true] TCP Connection keep alive enabled * @param {number} [options.socketOptions.keepAliveInitialDelay=120000] The number of milliseconds to wait before initiating keepAlive on the TCP socket * @param {number} [options.socketOptions.connectTimeoutMS=10000] How long to wait for a connection to be established before timing out * @param {number} [options.socketOptions.socketTimeoutMS=360000] How long a send or receive on a socket can take before timing out * @param {boolean} [options.domainsEnabled=false] Enable the wrapping of the callback in the current domain, disabled by default to avoid perf hit. * @param {number} [options.maxStalenessSeconds=undefined] The max staleness to secondary reads (values under 10 seconds cannot be guaranteed); * @param {boolean} [options.monitorCommands=false] Enable command monitoring for this topology * @fires ReplSet#connect * @fires ReplSet#ha * @fires ReplSet#joined * @fires ReplSet#left * @fires ReplSet#fullsetup * @fires ReplSet#open * @fires ReplSet#close * @fires ReplSet#error * @fires ReplSet#timeout * @fires ReplSet#parseError * @fires ReplSet#commandStarted * @fires ReplSet#commandSucceeded * @fires ReplSet#commandFailed * @property {string} parserType the parser type used (c++ or js). * @return {ReplSet} a ReplSet instance. */ class ReplSet extends TopologyBase { constructor(servers, options) { super(); options = options || {}; var self = this; // Filter the options options = filterOptions(options, legalOptionNames); // Ensure all the instances are Server for (var i = 0; i < servers.length; i++) { if (!(servers[i] instanceof Server)) { throw MongoError.create({ message: 'all seed list instances must be of the Server type', driver: true }); } } // Stored options var storeOptions = { force: false, bufferMaxEntries: typeof options.bufferMaxEntries === 'number' ? options.bufferMaxEntries : MAX_JS_INT }; // Shared global store var store = options.store || new Store(self, storeOptions); // Build seed list var seedlist = servers.map(function(x) { return { host: x.host, port: x.port }; }); // Clone options var clonedOptions = mergeOptions( {}, { disconnectHandler: store, cursorFactory: Cursor, reconnect: false, emitError: typeof options.emitError === 'boolean' ? options.emitError : true, size: typeof options.poolSize === 'number' ? options.poolSize : 5, monitorCommands: typeof options.monitorCommands === 'boolean' ? options.monitorCommands : false } ); // Translate any SSL options and other connectivity options clonedOptions = translateOptions(clonedOptions, options); // Socket options var socketOptions = options.socketOptions && Object.keys(options.socketOptions).length > 0 ? options.socketOptions : options; // Translate all the options to the core types clonedOptions = translateOptions(clonedOptions, socketOptions); // Create the ReplSet var coreTopology = new CReplSet(seedlist, clonedOptions); // Listen to reconnect event coreTopology.on('reconnect', function() { self.emit('reconnect'); store.execute(); }); // Internal state this.s = { // Replicaset coreTopology: coreTopology, // Server capabilities sCapabilities: null, // Debug tag tag: options.tag, // Store options storeOptions: storeOptions, // Cloned options clonedOptions: clonedOptions, // Store store: store, // Options options: options, // Server Session Pool sessionPool: null, // Active client sessions sessions: new Set(), // Promise library promiseLibrary: options.promiseLibrary || Promise }; // Debug if (clonedOptions.debug) { // Last ismaster Object.defineProperty(this, 'replset', { enumerable: true, get: function() { return coreTopology; } }); } } // Connect method connect(_options, callback) { var self = this; if ('function' === typeof _options) (callback = _options), (_options = {}); if (_options == null) _options = {}; if (!('function' === typeof callback)) callback = null; _options = Object.assign({}, this.s.clonedOptions, _options); self.s.options = _options; // Update bufferMaxEntries self.s.storeOptions.bufferMaxEntries = typeof _options.bufferMaxEntries === 'number' ? _options.bufferMaxEntries : -1; // Actual handler var errorHandler = function(event) { return function(err) { if (event !== 'error') { self.emit(event, err); } }; }; // Clear out all the current handlers left over var events = [ 'timeout', 'error', 'close', 'serverOpening', 'serverDescriptionChanged', 'serverHeartbeatStarted', 'serverHeartbeatSucceeded', 'serverHeartbeatFailed', 'serverClosed', 'topologyOpening', 'topologyClosed', 'topologyDescriptionChanged', 'commandStarted', 'commandSucceeded', 'commandFailed', 'joined', 'left', 'ping', 'ha' ]; events.forEach(function(e) { self.s.coreTopology.removeAllListeners(e); }); // relay the event var relay = function(event) { return function(t, server) { self.emit(event, t, server); }; }; // Replset events relay var replsetRelay = function(event) { return function(t, server) { self.emit(event, t, server.lastIsMaster(), server); }; }; // Relay ha var relayHa = function(t, state) { self.emit('ha', t, state); if (t === 'start') { self.emit('ha_connect', t, state); } else if (t === 'end') { self.emit('ha_ismaster', t, state); } }; // Set up serverConfig listeners self.s.coreTopology.on('joined', replsetRelay('joined')); self.s.coreTopology.on('left', relay('left')); self.s.coreTopology.on('ping', relay('ping')); self.s.coreTopology.on('ha', relayHa); // Set up SDAM listeners self.s.coreTopology.on('serverDescriptionChanged', relay('serverDescriptionChanged')); self.s.coreTopology.on('serverHeartbeatStarted', relay('serverHeartbeatStarted')); self.s.coreTopology.on('serverHeartbeatSucceeded', relay('serverHeartbeatSucceeded')); self.s.coreTopology.on('serverHeartbeatFailed', relay('serverHeartbeatFailed')); self.s.coreTopology.on('serverOpening', relay('serverOpening')); self.s.coreTopology.on('serverClosed', relay('serverClosed')); self.s.coreTopology.on('topologyOpening', relay('topologyOpening')); self.s.coreTopology.on('topologyClosed', relay('topologyClosed')); self.s.coreTopology.on('topologyDescriptionChanged', relay('topologyDescriptionChanged')); self.s.coreTopology.on('commandStarted', relay('commandStarted')); self.s.coreTopology.on('commandSucceeded', relay('commandSucceeded')); self.s.coreTopology.on('commandFailed', relay('commandFailed')); self.s.coreTopology.on('fullsetup', function() { self.emit('fullsetup', self, self); }); self.s.coreTopology.on('all', function() { self.emit('all', null, self); }); // Connect handler var connectHandler = function() { // Set up listeners self.s.coreTopology.once('timeout', errorHandler('timeout')); self.s.coreTopology.once('error', errorHandler('error')); self.s.coreTopology.once('close', errorHandler('close')); // Emit open event self.emit('open', null, self); // Return correctly try { callback(null, self); } catch (err) { process.nextTick(function() { throw err; }); } }; // Error handler var connectErrorHandler = function() { return function(err) { ['timeout', 'error', 'close'].forEach(function(e) { self.s.coreTopology.removeListener(e, connectErrorHandler); }); self.s.coreTopology.removeListener('connect', connectErrorHandler); // Destroy the replset self.s.coreTopology.destroy(); // Try to callback try { callback(err); } catch (err) { if (!self.s.coreTopology.isConnected()) process.nextTick(function() { throw err; }); } }; }; // Set up listeners self.s.coreTopology.once('timeout', connectErrorHandler('timeout')); self.s.coreTopology.once('error', connectErrorHandler('error')); self.s.coreTopology.once('close', connectErrorHandler('close')); self.s.coreTopology.once('connect', connectHandler); // Start connection self.s.coreTopology.connect(_options); } close(forceClosed, callback) { ['timeout', 'error', 'close', 'joined', 'left'].forEach(e => this.removeAllListeners(e)); super.close(forceClosed, callback); } } Object.defineProperty(ReplSet.prototype, 'haInterval', { enumerable: true, get: function() { return this.s.coreTopology.s.haInterval; } }); /** * A replset connect event, used to verify that the connection is up and running * * @event ReplSet#connect * @type {ReplSet} */ /** * The replset high availability event * * @event ReplSet#ha * @type {function} * @param {string} type The stage in the high availability event (start|end) * @param {boolean} data.norepeat This is a repeating high availability process or a single execution only * @param {number} data.id The id for this high availability request * @param {object} data.state An object containing the information about the current replicaset */ /** * A server member left the replicaset * * @event ReplSet#left * @type {function} * @param {string} type The type of member that left (primary|secondary|arbiter) * @param {Server} server The server object that left */ /** * A server member joined the replicaset * * @event ReplSet#joined * @type {function} * @param {string} type The type of member that joined (primary|secondary|arbiter) * @param {Server} server The server object that joined */ /** * ReplSet open event, emitted when replicaset can start processing commands. * * @event ReplSet#open * @type {Replset} */ /** * ReplSet fullsetup event, emitted when all servers in the topology have been connected to. * * @event ReplSet#fullsetup * @type {Replset} */ /** * ReplSet close event * * @event ReplSet#close * @type {object} */ /** * ReplSet error event, emitted if there is an error listener. * * @event ReplSet#error * @type {MongoError} */ /** * ReplSet timeout event * * @event ReplSet#timeout * @type {object} */ /** * ReplSet parseError event * * @event ReplSet#parseError * @type {object} */ /** * An event emitted indicating a command was started, if command monitoring is enabled * * @event ReplSet#commandStarted * @type {object} */ /** * An event emitted indicating a command succeeded, if command monitoring is enabled * * @event ReplSet#commandSucceeded * @type {object} */ /** * An event emitted indicating a command failed, if command monitoring is enabled * * @event ReplSet#commandFailed * @type {object} */ module.exports = ReplSet;