diff options
Diffstat (limited to 'node_modules/mongodb-core/lib/topologies/replset.js')
| -rw-r--r-- | node_modules/mongodb-core/lib/topologies/replset.js | 1278 |
1 files changed, 0 insertions, 1278 deletions
diff --git a/node_modules/mongodb-core/lib/topologies/replset.js b/node_modules/mongodb-core/lib/topologies/replset.js deleted file mode 100644 index dc73d59..0000000 --- a/node_modules/mongodb-core/lib/topologies/replset.js +++ /dev/null @@ -1,1278 +0,0 @@ -"use strict" - -var inherits = require('util').inherits, - f = require('util').format, - EventEmitter = require('events').EventEmitter, - BSON = require('bson').native().BSON, - ReadPreference = require('./read_preference'), - BasicCursor = require('../cursor'), - Logger = require('../connection/logger'), - debugOptions = require('../connection/utils').debugOptions, - MongoError = require('../error'), - Server = require('./server'), - ReplSetState = require('./replset_state'), - assign = require('./shared').assign; - -var MongoCR = require('../auth/mongocr') - , X509 = require('../auth/x509') - , Plain = require('../auth/plain') - , GSSAPI = require('../auth/gssapi') - , SSPI = require('../auth/sspi') - , ScramSHA1 = require('../auth/scram'); - -// -// States -var DISCONNECTED = 'disconnected'; -var CONNECTING = 'connecting'; -var CONNECTED = 'connected'; -var DESTROYED = 'destroyed'; - -function stateTransition(self, newState) { - var legalTransitions = { - 'disconnected': [CONNECTING, DESTROYED, DISCONNECTED], - 'connecting': [CONNECTING, DESTROYED, CONNECTED, DISCONNECTED], - 'connected': [CONNECTED, DISCONNECTED, 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 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.cert] SSL Certificate binary buffer - * @param {Buffer} [options.key] SSL Key file binary buffer - * @param {string} [options.passphrase] SSL Certificate pass phrase - * @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 {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 MongoS proxy selection - * @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 - */ -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: assign({}, options), - // BSON instance - bson: options.bson || new BSON(), - // 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, - 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 - } - - // 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 || { - 'mongocr': new MongoCR(this.s.bson), 'x509': new X509(this.s.bson) - , 'plain': new Plain(this.s.bson), 'gssapi': new GSSAPI(this.s.bson) - , 'sspi': new SSPI(this.s.bson), 'scram-sha-1': new ScramSHA1(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; -} - -inherits(ReplSet, EventEmitter); - -Object.defineProperty(ReplSet.prototype, 'type', { - enumerable:true, get: function() { return 'replset'; } -}); - -function attemptReconnect(self) { - self.haTimeoutId = setTimeout(function() { - if(self.state == DESTROYED) return; - - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('attemptReconnect for replset with id %s', self.id)); - } - - // Get all known hosts - var keys = Object.keys(self.s.replicaSetState.set); - var servers = keys.map(function(x) { - return new Server(assign({}, self.s.options, { - host: x.split(':')[0], port: parseInt(x.split(':')[1], 10) - }, { - authProviders: self.authProviders, reconnect:false, monitoring: false, inTopology: true - })); - }); - - // Create the list of servers - self.s.connectingServers = servers.slice(0); - - // Handle all events coming from servers - function _handleEvent(self, event) { - return function(err) { - // Destroy the instance - if(self.state == DESTROYED) { - return this.destroy(); - } - - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('attemptReconnect for replset with id %s using server %s ended with event %s', self.id, this.name, event)); - } - - // Check if we are done - function done() { - // Done with the reconnection attempt - if(self.s.connectingServers.length == 0) { - if(self.state == DESTROYED) return; - - // Do we have a primary - if(self.s.replicaSetState.hasPrimary()) { - // If we have a primary and a disconnect handler, execute - // buffered operations - if(self.s.replicaSetState.hasPrimaryAndSecondary() && self.s.disconnectHandler) { - self.s.disconnectHandler.execute(); - } - - // Connect any missing servers - connectNewServers(self, self.s.replicaSetState.unknownServers, function(err, cb) { - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('attemptReconnect for replset with id successful resuming topologyMonitor', self.id)); - } - // Go back to normal topology monitoring - topologyMonitor(self); - }); - } else { - if(self.listeners("close").length > 0) { - self.emit('close', self); - } - - attemptReconnect(self); - } - } - } - - // Remove the server from our list - for(var i = 0; i < self.s.connectingServers.length; i++) { - if(self.s.connectingServers[i].equals(this)) { - self.s.connectingServers.splice(i, 1); - } - } - - // Keep reference to server - var _self = this; - - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('attemptReconnect in replset with id %s for', self.id)); - } - - // Connect and not authenticating - if(event == 'connect' && !self.authenticating) { - if(self.state == DESTROYED) { - return _self.destroy(); - } - - // Update the replicaset state - if(self.s.replicaSetState.update(_self)) { - // 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')); - } else { - _self.destroy(); - } - - done(); - } else if(event == 'connect' && self.authenticating) { - this.destroy(); - } - - done(); - } - } - - // Index used to interleaf the server connects, avoiding - // runtime issues on io constrained vm's - var timeoutInterval = 0; - - function connect(server, timeoutInterval) { - setTimeout(function() { - 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', function(e) { self.emit('serverOpening', e); }); - server.on('serverDescriptionChanged', function(e) { self.emit('serverDescriptionChanged', e); }); - server.on('serverClosed', function(e) { self.emit('serverClosed', e); }); - - server.connect(self.s.connectOptions); - }, timeoutInterval); - } - - // Connect all servers - while(servers.length > 0) { - connect(servers.shift(), timeoutInterval++); - } - }, self.s.minHeartbeatFrequencyMS); -} - -function connectNewServers(self, servers, callback) { - // Count lefts - var count = servers.length; - - // Handle events - var _handleEvent = function(self, event) { - return function(err, r) { - var _self = this; - count = count - 1; - - // Destroyed - if(self.state == DESTROYED) { - return this.destroy(); - } - - if(event == 'connect' && !self.authenticating) { - // Destroyed - if(self.state == DESTROYED) { - return _self.destroy(); - } - - var result = self.s.replicaSetState.update(_self); - // Update the state with the new server - if(result) { - // 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')); - } else { - _self.destroy(); - } - } else if(event == 'connect' && self.authenticating) { - 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) { - return; - } - - // Create a new server instance - var server = new Server(assign({}, self.s.options, { - host: _server.split(':')[0], - port: parseInt(_server.split(':')[1], 10) - }, { - authProviders: self.authProviders, reconnect:false, monitoring: false, inTopology: true - })); - - // 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', function(e) { self.emit('serverOpening', e); }); - server.on('serverDescriptionChanged', function(e) { self.emit('serverDescriptionChanged', e); }); - server.on('serverClosed', function(e) { self.emit('serverClosed', e); }); - server.connect(self.s.connectOptions); - }, i); - } - - // Create new instances - for(var i = 0; i < servers.length; i++) { - execute(servers[i], i); - } -} - -function topologyMonitor(self, options) { - options = options || {}; - - // Set momitoring timeout - self.haTimeoutId = setTimeout(function() { - if(self.state == DESTROYED) return; - - // If we have a primary and a disconnect handler, execute - // buffered operations - if(self.s.replicaSetState.hasPrimaryAndSecondary() && self.s.disconnectHandler) { - self.s.disconnectHandler.execute(); - } - - // Get the connectingServers - var connectingServers = self.s.replicaSetState.allServers(); - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('topologyMonitor in replset with id %s connected servers [%s]' - , self.id - , connectingServers.map(function(x) { - return x.name; - }))); - } - // Get the count - var count = connectingServers.length; - // If we have no servers connected - if(count == 0 && !options.haInterval) { - if(self.listeners("close").length > 0) { - self.emit('close', self); - } - - return attemptReconnect(self); - } - - // 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}, function(err, r) { - if(self.state == DESTROYED) { - _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, 'serverHearbeatFailed', { durationMS: latencyMS, failure: err, connectionId: _server.name }); - // Remove the server from the state - _self.s.replicaSetState.remove(_server); - } else { - // Update the server ismaster - _server.ismaster = r.result; - 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; - } - - // _server.lastIsMasterMS = !latencyMS; - _self.s.replicaSetState.update(_server); - - // Server heart beat event - emitSDAMEvent(self, 'serverHeartbeatSucceeded', { durationMS: latencyMS, reply: r.result, connectionId: _server.name }); - } - - cb(err, r); - }); - } - - // Connect any missing servers - function connectMissingServers() { - if(self.state == DESTROYED) return; - - // Attempt to connect to any unknown servers - connectNewServers(self, self.s.replicaSetState.unknownServers, function(err, cb) { - if(self.state == DESTROYED) return; - // Check if we have an options.haInterval (meaning it was triggered from connect) - if(options.haInterval) { - // Do we have a primary and secondary - if(self.state == CONNECTING - && self.s.replicaSetState.hasPrimaryAndSecondary()) { - // Transition to connected - stateTransition(self, CONNECTED); - // Update initial state - self.initialConnectState.connect = true; - self.initialConnectState.fullsetup = true; - self.initialConnectState.all = true; - // Emit fullsetup and all events - process.nextTick(function() { - self.emit('connect', self); - self.emit('fullsetup', self); - self.emit('all', self); - }); - } else if(self.state == CONNECTING - && self.s.replicaSetState.hasPrimary()) { - // Transition to connected - stateTransition(self, CONNECTED); - // Update initial state - self.initialConnectState.connect = true; - // Emit connected sign - process.nextTick(function() { - self.emit('connect', self); - }); - } else if(self.state == CONNECTING - && self.s.replicaSetState.hasSecondary() - && self.s.options.secondaryOnlyConnectionAllowed) { - // Transition to connected - stateTransition(self, CONNECTED); - // Update initial state - self.initialConnectState.connect = true; - // Emit connected sign - process.nextTick(function() { - self.emit('connect', self); - }); - } else if(self.state == CONNECTING) { - self.emit('error', new MongoError('no primary found in replicaset')); - // Destroy the topology - return self.destroy(); - } else if(self.state == CONNECTED - && self.s.replicaSetState.hasPrimaryAndSecondary() - && !self.initialConnectState.fullsetup) { - self.initialConnectState.fullsetup = true; - // Emit fullsetup and all events - process.nextTick(function() { - self.emit('fullsetup', self); - self.emit('all', self); - }); - } - } - - topologyMonitor(self); - }); - } - - // No connectingServers but unknown servers - if(connectingServers.length == 0 - && self.s.replicaSetState.unknownServers.length > 0 && options.haInterval) { - return connectMissingServers(); - } else if(connectingServers.length == 0 && options.haInterval) { - self.destroy(); - return self.emit('error', new MongoError('no valid replicaset members found')); - } - - // Ping all servers - for(var i = 0; i < connectingServers.length; i++) { - pingServer(self, connectingServers[i], function(err, r) { - count = count - 1; - - if(count == 0) { - connectMissingServers(); - } - }); - } - }, options.haInterval || self.s.haInterval) -} - -function handleEvent(self, event) { - return function(err) { - if(self.state == DESTROYED) return; - // Debug log - if(self.s.logger.isDebug()) { - self.s.logger.debug(f('handleEvent %s from server %s in replset with id %s', event, self.name, self.id)); - } - - self.s.replicaSetState.remove(this); - } -} - -function handleInitialConnectEvent(self, event) { - return function(err) { - // 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) { - return this.destroy(); - } - - // Check the type of server - if(event == 'connect') { - // Update the state - var result = self.s.replicaSetState.update(this); - if(result == true) { - // 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')); - } else if(result instanceof MongoError) { - this.destroy(); - self.destroy(); - return self.emit('error', result); - } else { - this.destroy(); - } - } else { - // Emit failure to connect - self.emit('failed', this); - // Remove from the state - self.s.replicaSetState.remove(this); - } - - // 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) { - 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 - self.s.replicaSetState.update(server); - // 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', function(e) { self.emit('serverOpening', e); }); - server.on('serverDescriptionChanged', function(e) { self.emit('serverDescriptionChanged', e); }); - server.on('serverClosed', function(e) { self.emit('serverClosed', e); }); - // 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(assign({}, self.s.options, x, { - authProviders: self.authProviders, reconnect:false, monitoring:false, inTopology: true - })); - }); - - // Emit the topology opening event - emitSDAMEvent(this, 'topologyOpening', { topologyId: this.id }); - - // Start all server connections - connectServers(self, servers); -} - -/** - * Destroy the server connection - * @method - */ -ReplSet.prototype.destroy = function() { - // Transition state - stateTransition(this, DESTROYED); - // Clear out any monitoring process - if(this.haTimeoutId) clearTimeout(this.haTimeoutId); - // Destroy the replicaset - this.s.replicaSetState.destroy(); - - // Destroy all connecting servers - this.s.connectingServers.forEach(function(x) { - x.destroy(); - }); - - // 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, DISCONNECTED); - - 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() { - return this.s.replicaSetState.primary - ? this.s.replicaSetState.primary.lastIsMaster() : null; -} - -/** - * 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.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; -} - -/** - * Get server - * @method - * @param {ReadPreference} [options.readPreference] Specify read preference if command supports it - * @return {Server} - */ -ReplSet.prototype.getServer = function(options) { - // Ensure we have no options - options = options || {}; - // Pick the right server baspickServerd on readPreference - // var server = pickServer(this, this.s, options.readPreference); - var server = this.s.replicaSetState.pickServer(options.readPreference); - if(this.s.debug) this.emit('pickedServer', options.readPreference, server); - return server; -} - -/** - * Get all connected servers - * @method - * @return {Server[]} - */ -ReplSet.prototype.getServers = function() { - return this.s.replicaSetState.allServers(); -} - -function basicReadPreferenceValidation(self, options) { - if(options.readPreference && !(options.readPreference instanceof ReadPreference)) { - throw new Error("readPreference must be an instance of ReadPreference"); - } -} - -// -// Execute write operation -var executeWriteOperation = function(self, op, ns, ops, options, callback) { - if(typeof options == 'function') callback = options, options = {}, options = options || {}; - // Ensure we have no options - options = options || {}; - - // No server returned we had an error - if(self.s.replicaSetState.primary == null) { - return callback(new MongoError("no primary server found")); - } - - // Execute the command - self.s.replicaSetState.primary[op](ns, ops, options, callback); -} - -/** - * 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 {opResultCallback} callback A callback function - */ -ReplSet.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.s.replicaSetState.hasPrimary() && this.s.disconnectHandler != null) { - return this.s.disconnectHandler.add('insert', ns, ops, options, callback); - } - - // Execute write operation - executeWriteOperation(this, '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 {opResultCallback} callback A callback function - */ -ReplSet.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.s.replicaSetState.hasPrimary() && this.s.disconnectHandler != null) { - return this.s.disconnectHandler.add('insert', ns, ops, options, callback); - } - - // Execute write operation - executeWriteOperation(this, '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 {opResultCallback} callback A callback function - */ -ReplSet.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.s.replicaSetState.hasPrimary() && this.s.disconnectHandler != null) { - return this.s.disconnectHandler.add('insert', ns, ops, options, callback); - } - - // Execute write operation - executeWriteOperation(this, 'remove', ns, ops, options, callback); -} - -/** - * 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 {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; - // Pick a server - // var server = pickServer(self, self.s, readPreference); - // var server = self.s.replicaSetState.pickServer(readPreference); - var server = this.s.replicaSetState.pickServer(readPreference); - if(!(server instanceof Server)) return callback(server); - if(self.s.debug) self.emit('pickedServer', ReadPreference.primary, server); - - // 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 && 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(f("no server found that matches the provided readPreference %s", readPreference))); - } - - // Execute the command - server.command(ns, cmd, options, callback); -} - -/** - * 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(); - - // 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(!self.s.replicaSetState.hasPrimary() && self.s.disconnectHandler != null) { - 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, r) { - 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) return callback(MongoError.create({ - 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; - // 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); - } - - // Now logout all the servers - var servers = this.s.replicaSetState.allServers(); - var count = servers.length; - if(count == 0) return callback(); - var errors = []; - - // Execute logout on all server instances - for(var i = 0; i < servers.length; i++) { - servers[i].logout(dbName, function(err) { - count = count - 1; - if(err) errors.push({name: server.name, err: err}); - - if(count == 0) { - // Do not block new operations - self.authenticating = false; - // If we have one or more errors - if(errors.length) return callback(MongoError.create({ - message: f('logout failed against db %s', dbName), errors: errors - }), false); - - // No errors - callback(); - } - }); - } -} - -/** - * Perform one or more remove operations - * @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.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 {opResultCallback} callback A callback function - */ -ReplSet.prototype.cursor = function(ns, cmd, cursorOptions) { - cursorOptions = cursorOptions || {}; - var FinalCursor = cursorOptions.cursorFactory || this.s.Cursor; - return new FinalCursor(this.s.bson, ns, cmd, cursorOptions, this, 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 serverHearbeatFailed SDAM event - * - * @event ReplSet#serverHearbeatFailed - * @type {object} - */ - -/** - * A topology serverHeartbeatSucceeded SDAM change event - * - * @event ReplSet#serverHeartbeatSucceeded - * @type {object} - */ - -module.exports = ReplSet; |