12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724 |
- 'use strict';
- const inherits = require('util').inherits;
- const f = require('util').format;
- const EventEmitter = require('events').EventEmitter;
- const ReadPreference = require('./read_preference');
- const BasicCursor = require('../cursor');
- const retrieveBSON = require('../connection/utils').retrieveBSON;
- const Logger = require('../connection/logger');
- const MongoError = require('../error').MongoError;
- const Server = require('./server');
- const ReplSetState = require('./replset_state');
- const clone = require('./shared').clone;
- const Timeout = require('./shared').Timeout;
- const Interval = require('./shared').Interval;
- 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 defaultAuthProviders = require('../auth/defaultAuthProviders').defaultAuthProviders;
- var BSON = retrieveBSON();
- //
- // 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.s.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 Replset instance
- * @class
- * @param {array} seedlist A list of seeds for the replicaset
- * @param {boolean} options.setName The Replicaset set name
- * @param {boolean} [options.secondaryOnlyConnectionAllowed=false] Allow connection to a secondary only replicaset
- * @param {number} [options.haInterval=10000] The High availability period for replicaset inquiry
- * @param {boolean} [options.emitError=false] Server will emit errors events
- * @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 {boolean} [options.noDelay=true] TCP Connection no delay
- * @param {number} [options.connectionTimeout=10000] TCP Connection timeout setting
- * @param {number} [options.socketTimeout=0] TCP Socket timeout setting
- * @param {boolean} [options.singleBufferSerializtion=true] Serialize into single buffer, trade of peak memory for serialization speed
- * @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 {number} [options.pingInterval=5000] Ping interval to check the response time to the different servers
- * @param {number} [options.localThresholdMS=15] Cutoff latency point in MS for Replicaset member selection
- * @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 {ReplSet} A cursor instance
- * @fires ReplSet#connect
- * @fires ReplSet#ha
- * @fires ReplSet#joined
- * @fires ReplSet#left
- * @fires ReplSet#failed
- * @fires ReplSet#fullsetup
- * @fires ReplSet#all
- * @fires ReplSet#error
- * @fires ReplSet#serverHeartbeatStarted
- * @fires ReplSet#serverHeartbeatSucceeded
- * @fires ReplSet#serverHeartbeatFailed
- * @fires ReplSet#topologyOpening
- * @fires ReplSet#topologyClosed
- * @fires ReplSet#topologyDescriptionChanged
- * @property {string} type the topology type.
- * @property {string} parserType the parser type used (c++ or js).
- */
- var ReplSet = function(seedlist, options) {
- var self = this;
- options = options || {};
- // Validate seedlist
- if (!Array.isArray(seedlist)) throw new MongoError('seedlist must be an array');
- // Validate list
- if (seedlist.length === 0) throw new MongoError('seedlist must contain at least one entry');
- // Validate entries
- seedlist.forEach(function(e) {
- if (typeof e.host !== 'string' || typeof e.port !== 'number')
- throw new MongoError('seedlist entry must contain a host and port');
- });
- // Add event listener
- EventEmitter.call(this);
- // Get replSet Id
- this.id = id++;
- // Get the localThresholdMS
- var localThresholdMS = options.localThresholdMS || 15;
- // Backward compatibility
- if (options.acceptableLatency) localThresholdMS = options.acceptableLatency;
- // Create a logger
- var logger = Logger('ReplSet', options);
- // 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,
- // Seedlist
- seedlist: seedlist,
- // Replicaset state
- replicaSetState: new ReplSetState({
- id: this.id,
- setName: options.setName,
- acceptableLatency: localThresholdMS,
- heartbeatFrequencyMS: options.haInterval ? options.haInterval : 10000,
- logger: logger
- }),
- // Current servers we are connecting to
- connectingServers: [],
- // Ha interval
- haInterval: options.haInterval ? options.haInterval : 10000,
- // Minimum heartbeat frequency used if we detect a server close
- minHeartbeatFrequencyMS: 500,
- // 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,
- // Client info
- clientInfo: createClientInfo(options),
- // Authentication context
- authenticationContexts: []
- };
- // Add handler for topology change
- this.s.replicaSetState.on('topologyDescriptionChanged', function(r) {
- self.emit('topologyDescriptionChanged', r);
- });
- // 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
- )
- );
- }
- // All the authProviders
- this.authProviders = options.authProviders || defaultAuthProviders(this.s.bson);
- // Add forwarding of events from state handler
- var types = ['joined', 'left'];
- types.forEach(function(x) {
- self.s.replicaSetState.on(x, function(t, s) {
- self.emit(x, t, s);
- });
- });
- // Connect stat
- this.initialConnectState = {
- connect: false,
- fullsetup: false,
- all: false
- };
- // Disconnected state
- this.state = DISCONNECTED;
- this.haTimeoutId = null;
- // Are we authenticating
- this.authenticating = false;
- // Last ismaster
- this.ismaster = null;
- // Contains the intervalId
- this.intervalIds = [];
- // Highest clusterTime seen in responses from the current deployment
- this.clusterTime = null;
- };
- inherits(ReplSet, EventEmitter);
- Object.assign(ReplSet.prototype, SessionMixins);
- Object.defineProperty(ReplSet.prototype, 'type', {
- enumerable: true,
- get: function() {
- return 'replset';
- }
- });
- Object.defineProperty(ReplSet.prototype, 'parserType', {
- enumerable: true,
- get: function() {
- return BSON.native ? 'c++' : 'js';
- }
- });
- Object.defineProperty(ReplSet.prototype, 'logicalSessionTimeoutMinutes', {
- enumerable: true,
- get: function() {
- return this.s.replicaSetState.logicalSessionTimeoutMinutes || null;
- }
- });
- function rexecuteOperations(self) {
- // If we have a primary and a disconnect handler, execute
- // buffered operations
- if (self.s.replicaSetState.hasPrimaryAndSecondary() && self.s.disconnectHandler) {
- self.s.disconnectHandler.execute();
- } else if (self.s.replicaSetState.hasPrimary() && self.s.disconnectHandler) {
- self.s.disconnectHandler.execute({ executePrimary: true });
- } else if (self.s.replicaSetState.hasSecondary() && self.s.disconnectHandler) {
- self.s.disconnectHandler.execute({ executeSecondary: true });
- }
- }
- function connectNewServers(self, servers, callback) {
- // Count lefts
- var count = servers.length;
- var error = null;
- // Handle events
- var _handleEvent = function(self, event) {
- return function(err) {
- var _self = this;
- count = count - 1;
- // Destroyed
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return this.destroy({ force: true });
- }
- if (event === 'connect' && !self.authenticating) {
- // Destroyed
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return _self.destroy({ force: true });
- }
- // Do we have authentication contexts that need to be applied
- applyAuthenticationContexts(self, _self, function() {
- // Destroy the instance
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return _self.destroy({ force: true });
- }
- // Update the state
- var result = self.s.replicaSetState.update(_self);
- // Update the state with the new server
- if (result) {
- // Primary lastIsMaster store it
- if (_self.lastIsMaster() && _self.lastIsMaster().ismaster) {
- self.ismaster = _self.lastIsMaster();
- }
- // 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'));
- // Enalbe the monitoring of the new server
- monitorServer(_self.lastIsMaster().me, self, {});
- // Rexecute any stalled operation
- rexecuteOperations(self);
- } else {
- _self.destroy({ force: true });
- }
- });
- } else if (event === 'connect' && self.authenticating) {
- this.destroy({ force: true });
- } else if (event === 'error') {
- error = err;
- }
- // Rexecute any stalled operation
- rexecuteOperations(self);
- // Are we done finish up callback
- if (count === 0) {
- callback(error);
- }
- };
- };
- // 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.split(':')[0],
- port: parseInt(_server.split(':')[1], 10),
- authProviders: self.authProviders,
- reconnect: false,
- monitoring: false,
- parent: self,
- clientInfo: clone(self.s.clientInfo)
- })
- );
- // 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'));
- // SDAM Monitoring events
- server.on('serverOpening', e => self.emit('serverOpening', e));
- server.on('serverDescriptionChanged', e => self.emit('serverDescriptionChanged', e));
- server.on('serverClosed', e => self.emit('serverClosed', e));
- // Command Monitoring events
- relayEvents(server, self, ['commandStarted', 'commandSucceeded', 'commandFailed']);
- server.connect(self.s.connectOptions);
- }, i);
- }
- // Create new instances
- for (var i = 0; i < servers.length; i++) {
- execute(servers[i], i);
- }
- }
- // Ping the server
- var pingServer = function(self, server, cb) {
- // Measure running time
- var start = new Date().getTime();
- // Emit the server heartbeat start
- emitSDAMEvent(self, 'serverHeartbeatStarted', { connectionId: server.name });
- // Execute ismaster
- // Set the socketTimeout for a monitoring message to a low number
- // Ensuring ismaster calls are timed out quickly
- server.command(
- 'admin.$cmd',
- {
- ismaster: true
- },
- {
- monitoring: true,
- socketTimeout: self.s.options.connectionTimeout || 2000
- },
- function(err, r) {
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- server.destroy({ force: true });
- return cb(err, r);
- }
- // Calculate latency
- var latencyMS = new Date().getTime() - start;
- // Set the last updatedTime
- var hrTime = process.hrtime();
- // Calculate the last update time
- server.lastUpdateTime = hrTime[0] * 1000 + Math.round(hrTime[1] / 1000);
- // 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
- });
- // Remove server from the state
- self.s.replicaSetState.remove(server);
- } else {
- // Update the server ismaster
- server.ismaster = r.result;
- // Check if we have a lastWriteDate convert it to MS
- // and store on the server instance for later use
- if (server.ismaster.lastWrite && server.ismaster.lastWrite.lastWriteDate) {
- server.lastWriteDate = server.ismaster.lastWrite.lastWriteDate.getTime();
- }
- // Do we have a brand new server
- if (server.lastIsMasterMS === -1) {
- server.lastIsMasterMS = latencyMS;
- } else if (server.lastIsMasterMS) {
- // After the first measurement, average RTT MUST be computed using an
- // exponentially-weighted moving average formula, with a weighting factor (alpha) of 0.2.
- // If the prior average is denoted old_rtt, then the new average (new_rtt) is
- // computed from a new RTT measurement (x) using the following formula:
- // alpha = 0.2
- // new_rtt = alpha * x + (1 - alpha) * old_rtt
- server.lastIsMasterMS = 0.2 * latencyMS + (1 - 0.2) * server.lastIsMasterMS;
- }
- if (self.s.replicaSetState.update(server)) {
- // Primary lastIsMaster store it
- if (server.lastIsMaster() && server.lastIsMaster().ismaster) {
- self.ismaster = server.lastIsMaster();
- }
- }
- // Server heart beat event
- emitSDAMEvent(self, 'serverHeartbeatSucceeded', {
- durationMS: latencyMS,
- reply: r.result,
- connectionId: server.name
- });
- }
- // Calculate the staleness for this server
- self.s.replicaSetState.updateServerMaxStaleness(server, self.s.haInterval);
- // Callback
- cb(err, r);
- }
- );
- };
- // Each server is monitored in parallel in their own timeout loop
- var monitorServer = function(host, self, options) {
- // If this is not the initial scan
- // Is this server already being monitoried, then skip monitoring
- if (!options.haInterval) {
- for (var i = 0; i < self.intervalIds.length; i++) {
- if (self.intervalIds[i].__host === host) {
- return;
- }
- }
- }
- // Get the haInterval
- var _process = options.haInterval ? Timeout : Interval;
- var _haInterval = options.haInterval ? options.haInterval : self.s.haInterval;
- // Create the interval
- var intervalId = new _process(function() {
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- // clearInterval(intervalId);
- intervalId.stop();
- return;
- }
- // Do we already have server connection available for this host
- var _server = self.s.replicaSetState.get(host);
- // Check if we have a known server connection and reuse
- if (_server) {
- // Ping the server
- return pingServer(self, _server, function(err) {
- if (err) {
- // NOTE: should something happen here?
- return;
- }
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- intervalId.stop();
- return;
- }
- // Filter out all called intervaliIds
- self.intervalIds = self.intervalIds.filter(function(intervalId) {
- return intervalId.isRunning();
- });
- // Initial sweep
- if (_process === Timeout) {
- if (
- self.state === CONNECTING &&
- ((self.s.replicaSetState.hasSecondary() &&
- self.s.options.secondaryOnlyConnectionAllowed) ||
- self.s.replicaSetState.hasPrimary())
- ) {
- self.state = CONNECTED;
- // Emit connected sign
- process.nextTick(function() {
- self.emit('connect', self);
- });
- // Start topology interval check
- topologyMonitor(self, {});
- }
- } else {
- if (
- self.state === DISCONNECTED &&
- ((self.s.replicaSetState.hasSecondary() &&
- self.s.options.secondaryOnlyConnectionAllowed) ||
- self.s.replicaSetState.hasPrimary())
- ) {
- self.state = CONNECTED;
- // Rexecute any stalled operation
- rexecuteOperations(self);
- // Emit connected sign
- process.nextTick(function() {
- self.emit('reconnect', self);
- });
- }
- }
- if (
- self.initialConnectState.connect &&
- !self.initialConnectState.fullsetup &&
- self.s.replicaSetState.hasPrimaryAndSecondary()
- ) {
- // Set initial connect state
- self.initialConnectState.fullsetup = true;
- self.initialConnectState.all = true;
- process.nextTick(function() {
- self.emit('fullsetup', self);
- self.emit('all', self);
- });
- }
- });
- }
- }, _haInterval);
- // Start the interval
- intervalId.start();
- // Add the intervalId host name
- intervalId.__host = host;
- // Add the intervalId to our list of intervalIds
- self.intervalIds.push(intervalId);
- };
- function topologyMonitor(self, options) {
- if (self.state === DESTROYED || self.state === UNREFERENCED) return;
- options = options || {};
- // Get the servers
- var servers = Object.keys(self.s.replicaSetState.set);
- // Get the haInterval
- var _process = options.haInterval ? Timeout : Interval;
- var _haInterval = options.haInterval ? options.haInterval : self.s.haInterval;
- if (_process === Timeout) {
- return connectNewServers(self, self.s.replicaSetState.unknownServers, function(err) {
- // Don't emit errors if the connection was already
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return;
- }
- if (!self.s.replicaSetState.hasPrimary() && !self.s.options.secondaryOnlyConnectionAllowed) {
- if (err) {
- return self.emit('error', err);
- }
- self.emit(
- 'error',
- new MongoError('no primary found in replicaset or invalid replica set name')
- );
- return self.destroy({ force: true });
- } else if (
- !self.s.replicaSetState.hasSecondary() &&
- self.s.options.secondaryOnlyConnectionAllowed
- ) {
- if (err) {
- return self.emit('error', err);
- }
- self.emit(
- 'error',
- new MongoError('no secondary found in replicaset or invalid replica set name')
- );
- return self.destroy({ force: true });
- }
- for (var i = 0; i < servers.length; i++) {
- monitorServer(servers[i], self, options);
- }
- });
- } else {
- for (var i = 0; i < servers.length; i++) {
- monitorServer(servers[i], self, options);
- }
- }
- // Run the reconnect process
- function executeReconnect(self) {
- return function() {
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return;
- }
- connectNewServers(self, self.s.replicaSetState.unknownServers, function() {
- var monitoringFrequencey = self.s.replicaSetState.hasPrimary()
- ? _haInterval
- : self.s.minHeartbeatFrequencyMS;
- // Create a timeout
- self.intervalIds.push(new Timeout(executeReconnect(self), monitoringFrequencey).start());
- });
- };
- }
- // Decide what kind of interval to use
- var intervalTime = !self.s.replicaSetState.hasPrimary()
- ? self.s.minHeartbeatFrequencyMS
- : _haInterval;
- self.intervalIds.push(new Timeout(executeReconnect(self), intervalTime).start());
- }
- function addServerToList(list, server) {
- for (var i = 0; i < list.length; i++) {
- if (list[i].name.toLowerCase() === server.name.toLowerCase()) return true;
- }
- list.push(server);
- }
- function handleEvent(self, event) {
- return function() {
- if (self.state === DESTROYED || self.state === UNREFERENCED) return;
- // Debug log
- if (self.s.logger.isDebug()) {
- self.s.logger.debug(
- f('handleEvent %s from server %s in replset with id %s', event, this.name, self.id)
- );
- }
- // Remove from the replicaset state
- self.s.replicaSetState.remove(this);
- // Are we in a destroyed state return
- if (self.state === DESTROYED || self.state === UNREFERENCED) return;
- // If no primary and secondary available
- if (
- !self.s.replicaSetState.hasPrimary() &&
- !self.s.replicaSetState.hasSecondary() &&
- self.s.options.secondaryOnlyConnectionAllowed
- ) {
- stateTransition(self, DISCONNECTED);
- } else if (!self.s.replicaSetState.hasPrimary()) {
- stateTransition(self, DISCONNECTED);
- }
- addServerToList(self.s.connectingServers, this);
- };
- }
- function applyAuthenticationContexts(self, server, callback) {
- if (self.s.authenticationContexts.length === 0) {
- return callback();
- }
- // Do not apply any auth contexts if it's an arbiter
- if (server.lastIsMaster() && server.lastIsMaster().arbiterOnly) {
- return callback();
- }
- // Copy contexts to ensure no modificiation in the middle of
- // auth process.
- var authContexts = self.s.authenticationContexts.slice(0);
- // Apply one of the contexts
- function applyAuth(authContexts, server, callback) {
- if (authContexts.length === 0) return callback();
- // Get the first auth context
- var authContext = authContexts.shift();
- // Copy the params
- var customAuthContext = authContext.slice(0);
- // Push our callback handler
- customAuthContext.push(function(/* err */) {
- applyAuth(authContexts, server, callback);
- });
- // Attempt authentication
- server.auth.apply(server, customAuthContext);
- }
- // Apply all auth contexts
- applyAuth(authContexts, server, callback);
- }
- function shouldTriggerConnect(self) {
- const isConnecting = self.state === CONNECTING;
- const hasPrimary = self.s.replicaSetState.hasPrimary();
- const hasSecondary = self.s.replicaSetState.hasSecondary();
- const secondaryOnlyConnectionAllowed = self.s.options.secondaryOnlyConnectionAllowed;
- const readPreferenceSecondary =
- self.s.connectOptions.readPreference &&
- self.s.connectOptions.readPreference.equals(ReadPreference.secondary);
- return (
- (isConnecting &&
- ((readPreferenceSecondary && hasSecondary) || (!readPreferenceSecondary && hasPrimary))) ||
- (hasSecondary && secondaryOnlyConnectionAllowed)
- );
- }
- function handleInitialConnectEvent(self, event) {
- return function() {
- var _this = this;
- // Debug log
- if (self.s.logger.isDebug()) {
- self.s.logger.debug(
- f(
- 'handleInitialConnectEvent %s from server %s in replset with id %s',
- event,
- this.name,
- self.id
- )
- );
- }
- // Destroy the instance
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return this.destroy({ force: true });
- }
- // Check the type of server
- if (event === 'connect') {
- // Do we have authentication contexts that need to be applied
- applyAuthenticationContexts(self, _this, function() {
- // Destroy the instance
- if (self.state === DESTROYED || self.state === UNREFERENCED) {
- return _this.destroy({ force: true });
- }
- // Update the state
- var result = self.s.replicaSetState.update(_this);
- if (result === true) {
- // Primary lastIsMaster store it
- if (_this.lastIsMaster() && _this.lastIsMaster().ismaster) {
- self.ismaster = _this.lastIsMaster();
- }
- // Debug log
- if (self.s.logger.isDebug()) {
- self.s.logger.debug(
- f(
- 'handleInitialConnectEvent %s from server %s in replset with id %s has state [%s]',
- event,
- _this.name,
- self.id,
- JSON.stringify(self.s.replicaSetState.set)
- )
- );
- }
- // Remove the handlers
- for (var 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'));
- // Do we have a primary or primaryAndSecondary
- if (shouldTriggerConnect(self)) {
- // We are connected
- self.state = CONNECTED;
- // Set initial connect state
- self.initialConnectState.connect = true;
- // Emit connect event
- process.nextTick(function() {
- self.emit('connect', self);
- });
- topologyMonitor(self, {});
- }
- } else if (result instanceof MongoError) {
- _this.destroy({ force: true });
- self.destroy({ force: true });
- return self.emit('error', result);
- } else {
- _this.destroy({ force: true });
- }
- });
- } else {
- // Emit failure to connect
- self.emit('failed', this);
- addServerToList(self.s.connectingServers, this);
- // Remove from the state
- self.s.replicaSetState.remove(this);
- }
- if (
- self.initialConnectState.connect &&
- !self.initialConnectState.fullsetup &&
- self.s.replicaSetState.hasPrimaryAndSecondary()
- ) {
- // Set initial connect state
- self.initialConnectState.fullsetup = true;
- self.initialConnectState.all = true;
- process.nextTick(function() {
- self.emit('fullsetup', self);
- self.emit('all', self);
- });
- }
- // Remove from the list from connectingServers
- for (var i = 0; i < self.s.connectingServers.length; i++) {
- if (self.s.connectingServers[i].equals(this)) {
- self.s.connectingServers.splice(i, 1);
- }
- }
- // Trigger topologyMonitor
- if (self.s.connectingServers.length === 0 && self.state === CONNECTING) {
- topologyMonitor(self, { haInterval: 1 });
- }
- };
- }
- function connectServers(self, servers) {
- // Update connectingServers
- self.s.connectingServers = self.s.connectingServers.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() {
- // Add the server to the state
- if (self.s.replicaSetState.update(server)) {
- // Primary lastIsMaster store it
- if (server.lastIsMaster() && server.lastIsMaster().ismaster) {
- self.ismaster = server.lastIsMaster();
- }
- }
- // 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'));
- // SDAM Monitoring events
- server.on('serverOpening', e => self.emit('serverOpening', e));
- server.on('serverDescriptionChanged', e => self.emit('serverDescriptionChanged', e));
- server.on('serverClosed', e => self.emit('serverClosed', e));
- // 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++);
- }
- }
- /**
- * Emit event if it exists
- * @method
- */
- function emitSDAMEvent(self, event, description) {
- if (self.listeners(event).length > 0) {
- self.emit(event, description);
- }
- }
- /**
- * Initiate server connect
- * @method
- * @param {array} [options.auth=null] Array of auth options to apply on connect
- */
- ReplSet.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) {
- return new Server(
- Object.assign({}, self.s.options, x, options, {
- authProviders: self.authProviders,
- reconnect: false,
- monitoring: false,
- parent: self,
- clientInfo: clone(self.s.clientInfo)
- })
- );
- });
- // Error out as high availbility interval must be < than socketTimeout
- if (
- this.s.options.socketTimeout > 0 &&
- this.s.options.socketTimeout <= this.s.options.haInterval
- ) {
- return self.emit(
- 'error',
- new MongoError(
- f(
- 'haInterval [%s] MS must be set to less than socketTimeout [%s] MS',
- this.s.options.haInterval,
- this.s.options.socketTimeout
- )
- )
- );
- }
- // Emit the topology opening event
- emitSDAMEvent(this, 'topologyOpening', { topologyId: this.id });
- // Start all server connections
- connectServers(self, servers);
- };
- /**
- * Destroy the server connection
- * @param {boolean} [options.force=false] Force destroy the pool
- * @method
- */
- ReplSet.prototype.destroy = function(options) {
- options = options || {};
- // Transition state
- stateTransition(this, DESTROYED);
- // Clear out any monitoring process
- if (this.haTimeoutId) clearTimeout(this.haTimeoutId);
- // Destroy the replicaset
- this.s.replicaSetState.destroy(options);
- // Clear out authentication contexts
- this.s.authenticationContexts = [];
- // Destroy all connecting servers
- this.s.connectingServers.forEach(function(x) {
- x.destroy(options);
- });
- // Clear out all monitoring
- for (var i = 0; i < this.intervalIds.length; i++) {
- this.intervalIds[i].stop();
- }
- // Reset list of intervalIds
- this.intervalIds = [];
- // Emit toplogy closing event
- emitSDAMEvent(this, 'topologyClosed', { topologyId: this.id });
- };
- /**
- * Unref all connections belong to this server
- * @method
- */
- ReplSet.prototype.unref = function() {
- // Transition state
- stateTransition(this, UNREFERENCED);
- this.s.replicaSetState.allServers().forEach(function(x) {
- x.unref();
- });
- clearTimeout(this.haTimeoutId);
- };
- /**
- * Returns the last known ismaster document for this server
- * @method
- * @return {object}
- */
- ReplSet.prototype.lastIsMaster = function() {
- // If secondaryOnlyConnectionAllowed and no primary but secondary
- // return the secondaries ismaster result.
- if (
- this.s.options.secondaryOnlyConnectionAllowed &&
- !this.s.replicaSetState.hasPrimary() &&
- this.s.replicaSetState.hasSecondary()
- ) {
- return this.s.replicaSetState.secondaries[0].lastIsMaster();
- }
- return this.s.replicaSetState.primary
- ? this.s.replicaSetState.primary.lastIsMaster()
- : this.ismaster;
- };
- /**
- * All raw connections
- * @method
- * @return {Connection[]}
- */
- ReplSet.prototype.connections = function() {
- var servers = this.s.replicaSetState.allServers();
- var connections = [];
- for (var i = 0; i < servers.length; i++) {
- connections = connections.concat(servers[i].connections());
- }
- return connections;
- };
- /**
- * Figure out if the server is connected
- * @method
- * @param {ReadPreference} [options.readPreference] Specify read preference if command supports it
- * @return {boolean}
- */
- ReplSet.prototype.isConnected = function(options) {
- options = options || {};
- // If we are authenticating signal not connected
- // To avoid interleaving of operations
- if (this.authenticating) return false;
- // If we specified a read preference check if we are connected to something
- // than can satisfy this
- if (options.readPreference && options.readPreference.equals(ReadPreference.secondary)) {
- return this.s.replicaSetState.hasSecondary();
- }
- if (options.readPreference && options.readPreference.equals(ReadPreference.primary)) {
- return this.s.replicaSetState.hasPrimary();
- }
- if (options.readPreference && options.readPreference.equals(ReadPreference.primaryPreferred)) {
- return this.s.replicaSetState.hasSecondary() || this.s.replicaSetState.hasPrimary();
- }
- if (options.readPreference && options.readPreference.equals(ReadPreference.secondaryPreferred)) {
- return this.s.replicaSetState.hasSecondary() || this.s.replicaSetState.hasPrimary();
- }
- if (this.s.options.secondaryOnlyConnectionAllowed && this.s.replicaSetState.hasSecondary()) {
- return true;
- }
- return this.s.replicaSetState.hasPrimary();
- };
- /**
- * Figure out if the replicaset instance was destroyed by calling destroy
- * @method
- * @return {boolean}
- */
- ReplSet.prototype.isDestroyed = function() {
- return this.state === DESTROYED;
- };
- /**
- * Selects a server
- *
- * @method
- * @param {function} selector Unused
- * @param {ReadPreference} [options.readPreference] Specify read preference if command supports it
- * @param {function} callback
- */
- ReplSet.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 = this.s.replicaSetState.pickServer(options.readPreference);
- if (this.s.debug) this.emit('pickedServer', options.readPreference, server);
- callback(null, server);
- };
- /**
- * Get all connected servers
- * @method
- * @return {Server[]}
- */
- ReplSet.prototype.getServers = function() {
- return this.s.replicaSetState.allServers();
- };
- //
- // Execute write operation
- 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;
- if (self.state === DESTROYED) {
- return callback(new MongoError(f('topology was destroyed')));
- }
- const willRetryWrite =
- !args.retrying &&
- !!options.retryWrites &&
- options.session &&
- isRetryableWritesSupported(self) &&
- !options.session.inTransaction();
- if (!self.s.replicaSetState.hasPrimary()) {
- if (self.s.disconnectHandler) {
- // Not connected but we have a disconnecthandler
- return self.s.disconnectHandler.add(op, ns, ops, options, callback);
- } else if (!willRetryWrite) {
- // No server returned we had an error
- return callback(new MongoError('no primary server found'));
- }
- }
- const handler = (err, result) => {
- if (!err) return callback(null, result);
- if (!isRetryableError(err)) {
- return callback(err);
- }
- if (willRetryWrite) {
- const newArgs = Object.assign({}, args, { retrying: true });
- return executeWriteOperation(newArgs, options, callback);
- }
- // Per SDAM, remove primary from replicaset
- if (self.s.replicaSetState.primary) {
- self.s.replicaSetState.remove(self.s.replicaSetState.primary, { force: true });
- }
- return callback(err);
- };
- if (callback.operationId) {
- handler.operationId = callback.operationId;
- }
- // increment and assign txnNumber
- if (willRetryWrite) {
- options.session.incrementTransactionNumber();
- options.willRetryWrite = willRetryWrite;
- }
- self.s.replicaSetState.primary[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
- */
- ReplSet.prototype.insert = function(ns, ops, options, callback) {
- // 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
- */
- ReplSet.prototype.update = function(ns, ops, options, callback) {
- // 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
- */
- ReplSet.prototype.remove = function(ns, ops, options, callback) {
- // 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
- */
- ReplSet.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;
- // Establish readPreference
- var readPreference = options.readPreference ? options.readPreference : ReadPreference.primary;
- // If the readPreference is primary and we have no primary, store it
- if (
- readPreference.preference === 'primary' &&
- !this.s.replicaSetState.hasPrimary() &&
- this.s.disconnectHandler != null
- ) {
- return this.s.disconnectHandler.add('command', ns, cmd, options, callback);
- } else if (
- readPreference.preference === 'secondary' &&
- !this.s.replicaSetState.hasSecondary() &&
- this.s.disconnectHandler != null
- ) {
- return this.s.disconnectHandler.add('command', ns, cmd, options, callback);
- } else if (
- readPreference.preference !== 'primary' &&
- !this.s.replicaSetState.hasSecondary() &&
- !this.s.replicaSetState.hasPrimary() &&
- this.s.disconnectHandler != null
- ) {
- return this.s.disconnectHandler.add('command', ns, cmd, options, callback);
- }
- // Pick a server
- var server = this.s.replicaSetState.pickServer(readPreference);
- // We received an error, return it
- if (!(server instanceof Server)) return callback(server);
- // Emit debug event
- if (self.s.debug) self.emit('pickedServer', ReadPreference.primary, server);
- // No server returned we had an error
- if (server == null) {
- return callback(
- new MongoError(
- f('no server found that matches the provided readPreference %s', readPreference)
- )
- );
- }
- 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({}, options, { retrying: true });
- return this.command(ns, cmd, newOptions, callback);
- }
- // Per SDAM, remove primary from replicaset
- if (this.s.replicaSetState.primary) {
- this.s.replicaSetState.remove(this.s.replicaSetState.primary, { force: true });
- }
- return callback(err);
- };
- // increment and assign txnNumber
- if (willRetryWrite) {
- options.session.incrementTransactionNumber();
- options.willRetryWrite = willRetryWrite;
- }
- // Execute the command
- server.command(ns, cmd, options, cb);
- };
- /**
- * Authenticate using a specified mechanism
- * @method
- * @param {string} mechanism The Auth mechanism we are invoking
- * @param {string} db The db we are invoking the mechanism against
- * @param {...object} param Parameters for the specific mechanism
- * @param {authResultCallback} callback A callback function
- */
- ReplSet.prototype.auth = function(mechanism, db) {
- var allArgs = Array.prototype.slice.call(arguments, 0).slice(0);
- var self = this;
- var args = Array.prototype.slice.call(arguments, 2);
- var callback = args.pop();
- var currentContextIndex = 0;
- // If we don't have the mechanism fail
- if (this.authProviders[mechanism] == null && mechanism !== 'default') {
- return callback(new MongoError(f('auth provider %s does not exist', mechanism)));
- }
- // Are we already authenticating, throw
- if (this.authenticating) {
- return callback(new MongoError('authentication or logout allready in process'));
- }
- // 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 (!this.isConnected() && self.s.disconnectHandler != null) {
- if (!self.s.replicaSetState.hasPrimary() && !self.s.options.secondaryOnlyConnectionAllowed) {
- return self.s.disconnectHandler.add('auth', db, allArgs, {}, callback);
- } else if (
- !self.s.replicaSetState.hasSecondary() &&
- self.s.options.secondaryOnlyConnectionAllowed
- ) {
- return self.s.disconnectHandler.add('auth', db, allArgs, {}, callback);
- }
- }
- // Set to authenticating
- this.authenticating = true;
- // All errors
- var errors = [];
- // Get all the servers
- var servers = this.s.replicaSetState.allServers();
- // No servers return
- if (servers.length === 0) {
- this.authenticating = false;
- callback(null, true);
- }
- // Authenticate
- function auth(server) {
- // Arguments without a callback
- var argsWithoutCallback = [mechanism, db].concat(args.slice(0));
- // Create arguments
- var finalArguments = argsWithoutCallback.concat([
- function(err) {
- count = count - 1;
- // Save all the errors
- if (err) errors.push({ name: server.name, err: err });
- // We are done
- if (count === 0) {
- // Auth is done
- self.authenticating = false;
- // Return the auth error
- if (errors.length) {
- // Remove the entry from the stored authentication contexts
- self.s.authenticationContexts.splice(currentContextIndex, 0);
- // Return error
- return callback(
- new MongoError({
- message: 'authentication fail',
- errors: errors
- }),
- false
- );
- }
- // Successfully authenticated session
- callback(null, self);
- }
- }
- ]);
- if (!server.lastIsMaster().arbiterOnly) {
- // Execute the auth only against non arbiter servers
- server.auth.apply(server, finalArguments);
- } else {
- // If we are authenticating against an arbiter just ignore it
- finalArguments.pop()(null);
- }
- }
- // Get total count
- var count = servers.length;
- // Save current context index
- currentContextIndex = this.s.authenticationContexts.length;
- // Store the auth context and return the last index
- this.s.authenticationContexts.push([mechanism, db].concat(args.slice(0)));
- // Authenticate against all servers
- while (servers.length > 0) {
- auth(servers.shift());
- }
- };
- /**
- * Logout from a database
- * @method
- * @param {string} db The db we are logging out from
- * @param {authResultCallback} callback A callback function
- */
- ReplSet.prototype.logout = function(dbName, callback) {
- var self = this;
- // Are we authenticating or logging out, throw
- if (this.authenticating) {
- throw new MongoError('authentication or logout allready in process');
- }
- // Ensure no new members are processed while logging out
- this.authenticating = true;
- // Remove from all auth providers (avoid any reaplication of the auth details)
- var providers = Object.keys(this.authProviders);
- for (var i = 0; i < providers.length; i++) {
- this.authProviders[providers[i]].logout(dbName);
- }
- // Clear out any contexts associated with the db
- self.s.authenticationContexts = self.s.authenticationContexts.filter(function(context) {
- return context[1] !== dbName;
- });
- // Now logout all the servers
- var servers = this.s.replicaSetState.allServers();
- var count = servers.length;
- if (count === 0) return callback();
- var errors = [];
- function logoutServer(_server, cb) {
- _server.logout(dbName, function(err) {
- if (err) errors.push({ name: _server.name, err: err });
- cb();
- });
- }
- // Execute logout on all server instances
- for (i = 0; i < servers.length; i++) {
- logoutServer(servers[i], function() {
- count = count - 1;
- if (count === 0) {
- // Do not block new operations
- self.authenticating = false;
- // If we have one or more errors
- if (errors.length)
- return callback(
- new MongoError({
- message: f('logout failed against db %s', dbName),
- errors: errors
- }),
- false
- );
- // No errors
- callback();
- }
- });
- }
- };
- /**
- * 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}
- */
- ReplSet.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);
- };
- /**
- * A replset connect event, used to verify that the connection is up and running
- *
- * @event ReplSet#connect
- * @type {ReplSet}
- */
- /**
- * A replset reconnect event, used to verify that the topology reconnected
- *
- * @event ReplSet#reconnect
- * @type {ReplSet}
- */
- /**
- * A replset fullsetup event, used to signal that all topology members have been contacted.
- *
- * @event ReplSet#fullsetup
- * @type {ReplSet}
- */
- /**
- * A replset all event, used to signal that all topology members have been contacted.
- *
- * @event ReplSet#all
- * @type {ReplSet}
- */
- /**
- * A replset failed event, used to signal that initial replset connection failed.
- *
- * @event ReplSet#failed
- * @type {ReplSet}
- */
- /**
- * 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
- */
- /**
- * A server opening SDAM monitoring event
- *
- * @event ReplSet#serverOpening
- * @type {object}
- */
- /**
- * A server closed SDAM monitoring event
- *
- * @event ReplSet#serverClosed
- * @type {object}
- */
- /**
- * A server description SDAM change monitoring event
- *
- * @event ReplSet#serverDescriptionChanged
- * @type {object}
- */
- /**
- * A topology open SDAM event
- *
- * @event ReplSet#topologyOpening
- * @type {object}
- */
- /**
- * A topology closed SDAM event
- *
- * @event ReplSet#topologyClosed
- * @type {object}
- */
- /**
- * A topology structure SDAM change event
- *
- * @event ReplSet#topologyDescriptionChanged
- * @type {object}
- */
- /**
- * A topology serverHeartbeatStarted SDAM event
- *
- * @event ReplSet#serverHeartbeatStarted
- * @type {object}
- */
- /**
- * A topology serverHeartbeatFailed SDAM event
- *
- * @event ReplSet#serverHeartbeatFailed
- * @type {object}
- */
- /**
- * A topology serverHeartbeatSucceeded SDAM change event
- *
- * @event ReplSet#serverHeartbeatSucceeded
- * @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;
|