Node.js MongoDB Driver API

Source: lib/replset.js

  1. "use strict";
  3. var EventEmitter = require('events').EventEmitter
  4.   , inherits = require('util').inherits
  5.   , f = require('util').format
  6.   , Server = require('./server')
  7.   , Cursor = require('./cursor')
  8.   , AggregationCursor = require('./aggregation_cursor')
  9.   , CommandCursor = require('./command_cursor')
  10.   , ReadPreference = require('./read_preference')
  11.   , MongoError = require('mongodb-core').MongoError
  12.   , ServerCapabilities = require('./topology_base').ServerCapabilities
  13.   , Store = require('./topology_base').Store
  14.   , Define = require('./metadata')
  15.   , CReplSet = require('mongodb-core').ReplSet
  16.   , CoreReadPreference = require('mongodb-core').ReadPreference
  17.   , MAX_JS_INT = require('./utils').MAX_JS_INT
  18.   , translateOptions = require('./utils').translateOptions
  19.   , filterOptions = require('./utils').filterOptions
  20.   , getReadPreference = require('./utils').getReadPreference
  21.   , mergeOptions = require('./utils').mergeOptions
  22.   , os = require('os');
  23. /**
  24.  * @fileOverview The **ReplSet** class is a class that represents a Replicaset topology and is
  25.  * used to construct connections.
  26.  *
  27.  * **ReplSet Should not be used, use MongoClient.connect**
  28.  * @example
  29.  * var Db = require('mongodb').Db,
  30.  *   ReplSet = require('mongodb').ReplSet,
  31.  *   Server = require('mongodb').Server,
  32.  *   test = require('assert');
  33.  * // Connect using ReplSet
  34.  * var server = new Server('localhost', 27017);
  35.  * var db = new Db('test', new ReplSet([server]));
  36.  * db.open(function(err, db) {
  37.  *   // Get an additional db
  38.  *   db.close();
  39.  * });
  40.  */
  42. // Allowed parameters
  43. var legalOptionNames = ['ha', 'haInterval', 'replicaSet', 'rs_name', 'secondaryAcceptableLatencyMS'
  44.   , 'connectWithNoPrimary', 'poolSize', 'ssl', 'checkServerIdentity', 'sslValidate'
  45.   , 'sslCA', 'sslCert', 'sslCRL', 'sslKey', 'sslPass', 'socketOptions', 'bufferMaxEntries'
  46.   , 'store', 'auto_reconnect', 'autoReconnect', 'emitError'
  47.   , 'keepAlive', 'noDelay', 'connectTimeoutMS', 'socketTimeoutMS', 'strategy', 'debug', 'family'
  48.   , 'loggerLevel', 'logger', 'reconnectTries', 'appname', 'domainsEnabled'
  49.   , 'servername', 'promoteLongs', 'promoteValues', 'promoteBuffers', 'maxStalenessSeconds'];
  51. // Get package.json variable
  52. var driverVersion = require('../package.json').version;
  53. var nodejsversion = f('Node.js %s, %s', process.version, os.endianness());
  54. var type = os.type();
  55. var name = process.platform;
  56. var architecture = process.arch;
  57. var release = os.release();
  59. /**
  60.  * Creates a new ReplSet instance
  61.  * @class
  62.  * @deprecated
  63.  * @param {Server[]} servers A seedlist of servers participating in the replicaset.
  64.  * @param {object} [options=null] Optional settings.
  65.  * @param {boolean} [options.ha=true] Turn on high availability monitoring.
  66.  * @param {number} [options.haInterval=10000] Time between each replicaset status check.
  67.  * @param {string} [options.replicaSet] The name of the replicaset to connect to.
  68.  * @param {number} [options.secondaryAcceptableLatencyMS=15] Sets the range of servers to pick when using NEAREST (lowest ping ms + the latency fence, ex: range of 1 to (1 + 15) ms)
  69.  * @param {boolean} [options.connectWithNoPrimary=false] Sets if the driver should connect even if no primary is available
  70.  * @param {number} [options.poolSize=5] Number of connections in the connection pool for each server instance, set to 5 as default for legacy reasons.
  71.  * @param {boolean} [options.ssl=false] Use ssl connection (needs to have a mongod server with ssl support)
  72.  * @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.
  73.  * @param {object} [options.sslValidate=true] Validate mongod server certificate against ca (needs to have a mongod server with ssl support, 2.4 or higher)
  74.  * @param {array} [options.sslCA=null] Array of valid certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher)
  75.  * @param {array} [options.sslCRL=null] Array of revocation certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher)
  76.  * @param {(Buffer|string)} [options.sslCert=null] String or buffer containing the certificate we wish to present (needs to have a mongod server with ssl support, 2.4 or higher)
  77.  * @param {(Buffer|string)} [options.sslKey=null] String or buffer containing the certificate private key we wish to present (needs to have a mongod server with ssl support, 2.4 or higher)
  78.  * @param {(Buffer|string)} [options.sslPass=null] String or buffer containing the certificate password (needs to have a mongod server with ssl support, 2.4 or higher)
  79.  * @param {string} [options.servername=null] String containing the server name requested via TLS SNI.
  80.  * @param {object} [options.socketOptions=null] Socket options
  81.  * @param {boolean} [options.socketOptions.noDelay=true] TCP Socket NoDelay option.
  82.  * @param {number} [options.socketOptions.keepAlive=0] TCP KeepAlive on the socket with a X ms delay before start.
  83.  * @param {number} [options.socketOptions.connectTimeoutMS=10000] TCP Connection timeout setting
  84.  * @param {number} [options.socketOptions.socketTimeoutMS=0] TCP Socket timeout setting
  85.  * @param {boolean} [options.domainsEnabled=false] Enable the wrapping of the callback in the current domain, disabled by default to avoid perf hit.
  86.  * @param {number} [options.maxStalenessSeconds=undefined] The max staleness to secondary reads (values under 10 seconds cannot be guaranteed);
  87.  * @fires ReplSet#connect
  88.  * @fires ReplSet#ha
  89.  * @fires ReplSet#joined
  90.  * @fires ReplSet#left
  91.  * @fires ReplSet#fullsetup
  92.  * @fires ReplSet#open
  93.  * @fires ReplSet#close
  94.  * @fires ReplSet#error
  95.  * @fires ReplSet#timeout
  96.  * @fires ReplSet#parseError
  97.  * @property {string} parserType the parser type used (c++ or js).
  98.  * @return {ReplSet} a ReplSet instance.
  99.  */
  100. var ReplSet = function(servers, options) {
  101.   if(!(this instanceof ReplSet)) return new ReplSet(servers, options);
  102.   options = options || {};
  103.   var self = this;
  104.   // Set up event emitter
  105.   EventEmitter.call(this);
  107.   // Filter the options
  108.   options = filterOptions(options, legalOptionNames);
  110.   // Ensure all the instances are Server
  111.   for(var i = 0; i < servers.length; i++) {
  112.     if(!(servers[i] instanceof Server)) {
  113.       throw MongoError.create({message: "all seed list instances must be of the Server type", driver:true});
  114.     }
  115.   }
  117.   // Stored options
  118.   var storeOptions = {
  119.       force: false
  120.     , bufferMaxEntries: typeof options.bufferMaxEntries == 'number' ? options.bufferMaxEntries : MAX_JS_INT
  121.   }
  123.   // Shared global store
  124.   var store = options.store || new Store(self, storeOptions);
  126.   // Build seed list
  127.   var seedlist = servers.map(function(x) {
  128.     return {host: x.host, port: x.port}
  129.   });
  131.   // Clone options
  132.   var clonedOptions = mergeOptions({}, {
  133.     disconnectHandler: store,
  134.     cursorFactory: Cursor,
  135.     reconnect: false,
  136.     emitError: typeof options.emitError == 'boolean' ? options.emitError : true,
  137.     size: typeof options.poolSize == 'number' ? options.poolSize : 5
  138.   });
  140.   // Translate any SSL options and other connectivity options
  141.   clonedOptions = translateOptions(clonedOptions, options);
  143.   // Socket options
  144.   var socketOptions = options.socketOptions && Object.keys(options.socketOptions).length > 0
  145.     ? options.socketOptions : options;
  147.   // Translate all the options to the mongodb-core ones
  148.   clonedOptions = translateOptions(clonedOptions, socketOptions);
  149.   if(typeof clonedOptions.keepAlive == 'number') {
  150.     clonedOptions.keepAliveInitialDelay = clonedOptions.keepAlive;
  151.     clonedOptions.keepAlive = clonedOptions.keepAlive > 0;
  152.   }
  154.   // Client info
  155.   this.clientInfo = {
  156.     driver: {
  157.       name: "nodejs",
  158.       version: driverVersion
  159.     },
  160.     os: {
  161.       type: type,
  162.       name: name,
  163.       architecture: architecture,
  164.       version: release
  165.     },
  166.     platform: nodejsversion
  167.   }
  169.   // Build default client information
  170.   clonedOptions.clientInfo = this.clientInfo;
  171.   // Do we have an application specific string
  172.   if(options.appname) {
  173.     clonedOptions.clientInfo.application = { name: options.appname };
  174.   }
  176.   // Create the ReplSet
  177.   var replset = new CReplSet(seedlist, clonedOptions);
  179.   // Listen to reconnect event
  180.   replset.on('reconnect', function() {
  181.     self.emit('reconnect');
  182.     store.execute();
  183.   });
  185.   // Internal state
  186.   this.s = {
  187.     // Replicaset
  188.     replset: replset
  189.     // Server capabilities
  190.     , sCapabilities: null
  191.     // Debug tag
  192.     , tag: options.tag
  193.     // Store options
  194.     , storeOptions: storeOptions
  195.     // Cloned options
  196.     , clonedOptions: clonedOptions
  197.     // Store
  198.     , store: store
  199.     // Options
  200.     , options: options
  201.   }
  203.   // Debug
  204.   if(clonedOptions.debug) {
  205.     // Last ismaster
  206.     Object.defineProperty(this, 'replset', {
  207.       enumerable:true, get: function() { return replset; }
  208.     });
  209.   }
  210. }
  212. /**
  213.  * @ignore
  214.  */
  215. inherits(ReplSet, EventEmitter);
  217. // Last ismaster
  218. Object.defineProperty(ReplSet.prototype, 'isMasterDoc', {
  219.   enumerable:true, get: function() { return this.s.replset.lastIsMaster(); }
  220. });
  222. Object.defineProperty(ReplSet.prototype, 'parserType', {
  223.   enumerable:true, get: function() {
  224.     return this.s.replset.parserType;
  225.   }
  226. });
  228. // BSON property
  229. Object.defineProperty(ReplSet.prototype, 'bson', {
  230.   enumerable: true, get: function() {
  231.     return this.s.replset.s.bson;
  232.   }
  233. });
  235. Object.defineProperty(ReplSet.prototype, 'haInterval', {
  236.   enumerable:true, get: function() { return this.s.replset.s.haInterval; }
  237. });
  239. var define = ReplSet.define = new Define('ReplSet', ReplSet, false);
  241. // Ensure the right read Preference object
  242. var translateReadPreference = function(options) {
  243.   if(typeof options.readPreference == 'string') {
  244.     options.readPreference = new CoreReadPreference(options.readPreference);
  245.   } else if(options.readPreference instanceof ReadPreference) {
  246.     options.readPreference = new CoreReadPreference(options.readPreference.mode
  247.       , options.readPreference.tags, {maxStalenessSeconds: options.readPreference.maxStalenessSeconds});
  248.   }
  250.   return options;
  251. }
  253. // Connect method
  254. ReplSet.prototype.connect = function(db, _options, callback) {
  255.   var self = this;
  256.   if('function' === typeof _options) callback = _options, _options = {};
  257.   if(_options == null) _options = {};
  258.   if(!('function' === typeof callback)) callback = null;
  259.   self.s.options = _options;
  261.   // Update bufferMaxEntries
  262.   self.s.storeOptions.bufferMaxEntries = db.bufferMaxEntries;
  264.   // Actual handler
  265.   var errorHandler = function(event) {
  266.     return function(err) {
  267.       if(event != 'error') {
  268.         self.emit(event, err);
  269.       }
  270.     }
  271.   }
  273.   // Clear out all the current handlers left over
  274.   var events = ["timeout", "error", "close", 'serverOpening', 'serverDescriptionChanged', 'serverHeartbeatStarted',
  275.     'serverHeartbeatSucceeded', 'serverHeartbeatFailed', 'serverClosed', 'topologyOpening',
  276.     'topologyClosed', 'topologyDescriptionChanged', 'joined', 'left', 'ping', 'ha'];
  277.   events.forEach(function(e) {
  278.     self.s.replset.removeAllListeners(e);
  279.   });
  281.   // relay the event
  282.   var relay = function(event) {
  283.     return function(t, server) {
  284.       self.emit(event, t, server);
  285.     }
  286.   }
  288.   // Replset events relay
  289.   var replsetRelay = function(event) {
  290.     return function(t, server) {
  291.       self.emit(event, t, server.lastIsMaster(), server);
  292.     }
  293.   }
  295.   // Relay ha
  296.   var relayHa = function(t, state) {
  297.     self.emit('ha', t, state);
  299.     if(t == 'start') {
  300.       self.emit('ha_connect', t, state);
  301.     } else if(t == 'end') {
  302.       self.emit('ha_ismaster', t, state);
  303.     }
  304.   }
  306.   // Set up serverConfig listeners
  307.   self.s.replset.on('joined', replsetRelay('joined'));
  308.   self.s.replset.on('left', relay('left'));
  309.   self.s.replset.on('ping', relay('ping'));
  310.   self.s.replset.on('ha', relayHa);
  312.   // Set up SDAM listeners
  313.   self.s.replset.on('serverDescriptionChanged', relay('serverDescriptionChanged'));
  314.   self.s.replset.on('serverHeartbeatStarted', relay('serverHeartbeatStarted'));
  315.   self.s.replset.on('serverHeartbeatSucceeded', relay('serverHeartbeatSucceeded'));
  316.   self.s.replset.on('serverHeartbeatFailed', relay('serverHeartbeatFailed'));
  317.   self.s.replset.on('serverOpening', relay('serverOpening'));
  318.   self.s.replset.on('serverClosed', relay('serverClosed'));
  319.   self.s.replset.on('topologyOpening', relay('topologyOpening'));
  320.   self.s.replset.on('topologyClosed', relay('topologyClosed'));
  321.   self.s.replset.on('topologyDescriptionChanged', relay('topologyDescriptionChanged'));
  323.   self.s.replset.on('fullsetup', function() {
  324.     self.emit('fullsetup', self, self);
  325.   });
  327.   self.s.replset.on('all', function() {
  328.     self.emit('all', null, self);
  329.   });
  331.   // Connect handler
  332.   var connectHandler = function() {
  333.     // Set up listeners
  334.     self.s.replset.once('timeout', errorHandler('timeout'));
  335.     self.s.replset.once('error', errorHandler('error'));
  336.     self.s.replset.once('close', errorHandler('close'));
  338.     // Emit open event
  339.     self.emit('open', null, self);
  341.     // Return correctly
  342.     try {
  343.       callback(null, self);
  344.     } catch(err) {
  345.       process.nextTick(function() { throw err; })
  346.     }
  347.   }
  349.   // Error handler
  350.   var connectErrorHandler = function() {
  351.     return function(err) {
  352.       ['timeout', 'error', 'close'].forEach(function(e) {
  353.         self.s.replset.removeListener(e, connectErrorHandler);
  354.       });
  356.       self.s.replset.removeListener('connect', connectErrorHandler);
  357.       // Destroy the replset
  358.       self.s.replset.destroy();
  360.       // Try to callback
  361.       try {
  362.         callback(err);
  363.       } catch(err) {
  364.         if(!self.s.replset.isConnected())
  365.           process.nextTick(function() { throw err; })
  366.       }
  367.     }
  368.   }
  370.   // Set up listeners
  371.   self.s.replset.once('timeout', connectErrorHandler('timeout'));
  372.   self.s.replset.once('error', connectErrorHandler('error'));
  373.   self.s.replset.once('close', connectErrorHandler('close'));
  374.   self.s.replset.once('connect', connectHandler);
  376.   // Start connection
  377.   self.s.replset.connect(_options);
  378. }
  380. // Server capabilities
  381. ReplSet.prototype.capabilities = function() {
  382.   if(this.s.sCapabilities) return this.s.sCapabilities;
  383.   if(this.s.replset.lastIsMaster() == null) return null;
  384.   this.s.sCapabilities = new ServerCapabilities(this.s.replset.lastIsMaster());
  385.   return this.s.sCapabilities;
  386. }
  388. define.classMethod('capabilities', {callback: false, promise:false, returns: [ServerCapabilities]});
  390. // Command
  391. ReplSet.prototype.command = function(ns, cmd, options, callback) {
  392.   this.s.replset.command(ns, cmd, getReadPreference(options), callback);
  393. }
  395. define.classMethod('command', {callback: true, promise:false});
  397. // Insert
  398. ReplSet.prototype.insert = function(ns, ops, options, callback) {
  399.   this.s.replset.insert(ns, ops, options, callback);
  400. }
  402. define.classMethod('insert', {callback: true, promise:false});
  404. // Update
  405. ReplSet.prototype.update = function(ns, ops, options, callback) {
  406.   this.s.replset.update(ns, ops, options, callback);
  407. }
  409. define.classMethod('update', {callback: true, promise:false});
  411. // Remove
  412. ReplSet.prototype.remove = function(ns, ops, options, callback) {
  413.   this.s.replset.remove(ns, ops, options, callback);
  414. }
  416. define.classMethod('remove', {callback: true, promise:false});
  418. // Destroyed
  419. ReplSet.prototype.isDestroyed = function() {
  420.   return this.s.replset.isDestroyed();
  421. }
  423. // IsConnected
  424. ReplSet.prototype.isConnected = function(options) {
  425.   options = options || {};
  427.   // If we passed in a readPreference, translate to
  428.   // a CoreReadPreference instance
  429.   if(options.readPreference) {
  430.     options.readPreference = translateReadPreference(options.readPreference);
  431.   }
  433.   return this.s.replset.isConnected(options);
  434. }
  436. define.classMethod('isConnected', {callback: false, promise:false, returns: [Boolean]});
  438. // Insert
  439. ReplSet.prototype.cursor = function(ns, cmd, options) {
  440.   options = translateReadPreference(options);
  441.   options.disconnectHandler = this.s.store;
  442.   return this.s.replset.cursor(ns, cmd, options);
  443. }
  445. define.classMethod('cursor', {callback: false, promise:false, returns: [Cursor, AggregationCursor, CommandCursor]});
  447. ReplSet.prototype.lastIsMaster = function() {
  448.   return this.s.replset.lastIsMaster();
  449. }
  451. /**
  452.  * Unref all sockets
  453.  * @method
  454.  */
  455. ReplSet.prototype.unref = function() {
  456.   return this.s.replset.unref();
  457. }
  459. ReplSet.prototype.close = function(forceClosed) {
  460.   var self = this;
  461.   // Call destroy on the topology
  462.   this.s.replset.destroy({
  463.     force: typeof forceClosed == 'boolean' ? forceClosed : false,
  464.   });
  465.   // We need to wash out all stored processes
  466.   if(forceClosed == true) {
  467.     this.s.storeOptions.force = forceClosed;
  468.     this.s.store.flush();
  469.   }
  471.   var events = ['timeout', 'error', 'close', 'joined', 'left'];
  472.   events.forEach(function(e) {
  473.     self.removeAllListeners(e);
  474.   });
  475. }
  477. define.classMethod('close', {callback: false, promise:false});
  479. ReplSet.prototype.auth = function() {
  480.   var args = Array.prototype.slice.call(arguments, 0);
  481.   this.s.replset.auth.apply(this.s.replset, args);
  482. }
  484. define.classMethod('auth', {callback: true, promise:false});
  486. ReplSet.prototype.logout = function() {
  487.   var args = Array.prototype.slice.call(arguments, 0);
  488.   this.s.replset.logout.apply(this.s.replset, args);
  489. }
  491. define.classMethod('logout', {callback: true, promise:false});
  493. /**
  494.  * All raw connections
  495.  * @method
  496.  * @return {array}
  497.  */
  498. ReplSet.prototype.connections = function() {
  499.   return this.s.replset.connections();
  500. }
  502. define.classMethod('connections', {callback: false, promise:false, returns:[Array]});
  504. /**
  505.  * A replset connect event, used to verify that the connection is up and running
  506.  *
  507.  * @event ReplSet#connect
  508.  * @type {ReplSet}
  509.  */
  511. /**
  512.  * The replset high availability event
  513.  *
  514.  * @event ReplSet#ha
  515.  * @type {function}
  516.  * @param {string} type The stage in the high availability event (start|end)
  517.  * @param {boolean} data.norepeat This is a repeating high availability process or a single execution only
  518.  * @param {number} data.id The id for this high availability request
  519.  * @param {object} data.state An object containing the information about the current replicaset
  520.  */
  522. /**
  523.  * A server member left the replicaset
  524.  *
  525.  * @event ReplSet#left
  526.  * @type {function}
  527.  * @param {string} type The type of member that left (primary|secondary|arbiter)
  528.  * @param {Server} server The server object that left
  529.  */
  531. /**
  532.  * A server member joined the replicaset
  533.  *
  534.  * @event ReplSet#joined
  535.  * @type {function}
  536.  * @param {string} type The type of member that joined (primary|secondary|arbiter)
  537.  * @param {Server} server The server object that joined
  538.  */
  540. /**
  541.  * ReplSet open event, emitted when replicaset can start processing commands.
  542.  *
  543.  * @event ReplSet#open
  544.  * @type {Replset}
  545.  */
  547. /**
  548.  * ReplSet fullsetup event, emitted when all servers in the topology have been connected to.
  549.  *
  550.  * @event ReplSet#fullsetup
  551.  * @type {Replset}
  552.  */
  554. /**
  555.  * ReplSet close event
  556.  *
  557.  * @event ReplSet#close
  558.  * @type {object}
  559.  */
  561. /**
  562.  * ReplSet error event, emitted if there is an error listener.
  563.  *
  564.  * @event ReplSet#error
  565.  * @type {MongoError}
  566.  */
  568. /**
  569.  * ReplSet timeout event
  570.  *
  571.  * @event ReplSet#timeout
  572.  * @type {object}
  573.  */
  575. /**
  576.  * ReplSet parseError event
  577.  *
  578.  * @event ReplSet#parseError
  579.  * @type {object}
  580.  */
  582. module.exports = ReplSet;