Source: lib/replset.js

var EventEmitter = require('events').EventEmitter
  , inherits = require('util').inherits
  , f = require('util').format
  , Server = require('./server')
  , Mongos = require('./mongos')
  , Cursor = require('./cursor')
  , ReadPreference = require('./read_preference')
  , MongoCR = require('mongodb-core').MongoCR
  , MongoError = require('mongodb-core').MongoError
  , ServerCapabilities = require('./topology_base').ServerCapabilities
  , Store = require('./topology_base').Store
  , CServer = require('mongodb-core').Server
  , CReplSet = require('mongodb-core').ReplSet
  , CoreReadPreference = require('mongodb-core').ReadPreference
  , shallowClone = require('./utils').shallowClone;

/**
 * @fileOverview The **ReplSet** class is a class that represents a Replicaset topology and is
 * used to construct connections.
 * 
 * **ReplSet Should not be used, use MongoClient.connect**
 * @example
 * var Db = require('mongodb').Db,
 *   ReplSet = require('mongodb').ReplSet,
 *   Server = require('mongodb').Server,
 *   test = require('assert');
 * // Connect using ReplSet
 * var server = new Server('localhost', 27017);
 * var db = new Db('test', new ReplSet([server]));
 * db.open(function(err, db) {
 *   // Get an additional db
 *   db.close();
 * });
 */

/**
 * Creates a new ReplSet instance
 * @class
 * @deprecated
 * @param {Server[]} servers A seedlist of servers participating in the replicaset.
 * @param {object} [options=null] Optional settings.
 * @param {booelan} [options.ha=true] Turn on high availability monitoring.
 * @param {number} [options.haInterval=5000] Time between each replicaset status check.
 * @param {string} options.replicaSet The name of the replicaset to connect to.
 * @param {number} [options.secondaryAcceptableLatencyMS=15] Sets the range of servers to pick when using NEAREST (lowest ping ms + the latency fence, ex: range of 1 to (1 + 15) ms) 
 * @param {boolean} [options.connectWithNoPrimary=false] Sets if the driver should connect even if no primary is available
 * @param {number} [options.poolSize=5] Number of connections in the connection pool for each server instance, set to 5 as default for legacy reasons.
 * @param {boolean} [options.ssl=false] Use ssl connection (needs to have a mongod server with ssl support) 
 * @param {object} [options.sslValidate=false] Validate mongod server certificate against ca (needs to have a mongod server with ssl support, 2.4 or higher)
 * @param {array} [options.sslCA=null] Array of valid certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher)
 * @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)
 * @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)
 * @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)
 * @param {object} [options.socketOptions=null] Socket options 
 * @param {boolean} [options.socketOptions.noDelay=true] TCP Socket NoDelay option. 
 * @param {number} [options.socketOptions.keepAlive=0] TCP KeepAlive on the socket with a X ms delay before start. 
 * @param {number} [options.socketOptions.connectTimeoutMS=0] TCP Connection timeout setting 
 * @param {number} [options.socketOptions.socketTimeoutMS=0] TCP Socket timeout setting 
 * @fires ReplSet#connect
 * @fires ReplSet#ha
 * @fires ReplSet#joined
 * @fires ReplSet#left
 * @fires ReplSet#fullsetup
 * @fires ReplSet#open
 * @fires ReplSet#close
 * @fires ReplSet#error
 * @fires ReplSet#timeout
 * @fires ReplSet#parseError
 * @return {ReplSet} a ReplSet instance.
 */
var ReplSet = function(servers, options) {  
  if(!(this instanceof ReplSet)) return new ReplSet(servers, options);
  options = options || {};
  var self = this;

  // Ensure all the instances are Server
  for(var i = 0; i < servers.length; i++) {
    if(!(servers[i] instanceof Server)) {
      throw new MongoError("all seed list instances must be of the Server type");
    }
  }

  // Store option defaults
  var storeOptions = {
      force: false
    , bufferMaxEntries: -1
  }

  // Shared global store
  var store = options.store || new Store(self, storeOptions);

  // Set up event emitter
  EventEmitter.call(this);

  // Debug tag
  var tag = options.tag;

  // Build seed list
  var seedlist = servers.map(function(x) {
    return {host: x.host, port: x.port}
  });

  // Final options
  var finalOptions = shallowClone(options);

  // Default values
  finalOptions.size = options.poolSize || 5;  
  finalOptions.reconnect = typeof options.auto_reconnect == 'boolean' ? options.auto_reconnect : true;
  finalOptions.emitError = typeof options.emitError == 'boolean' ? options.emitError : true;
  finalOptions.cursorFactory = Cursor;

  // Add the store
  finalOptions.disconnectHandler = store;

  // Socket options passed down
  if(options.socketOptions) {
    if(options.socketOptions.connectTimeoutMS)
      finalOptions.connectionTimeout = options.socketOptions.connectTimeoutMS;
    if(options.socketOptions.socketTimeoutMS)
      finalOptions.socketTimeout = options.socketOptions.socketTimeoutMS;
  } 

  // Get the name
  var replicaSet = options.replicaSet || options.rs_name;

  // If no rs_name throw
  if(replicaSet == null) 
    throw new MongoError('replicaSet parameter must be set');

  // Set up options
  finalOptions.setName = replicaSet;

  // Are we running in debug mode
  var debug = typeof options.debug == 'boolean' ? options.debug : false;
  if(debug) {
    finalOptions.debug = debug;
  }

  // Map keep alive setting
  if(options.socketOptions && typeof options.socketOptions.keepAlive == 'number') {
    finalOptions.keepAlive = true;
    if(typeof options.socketOptions.keepAlive == 'number') {
      finalOptions.keepAliveInitialDelay = options.socketOptions.keepAlive;
    }
  }

  // Connection timeout
  if(options.socketOptions && typeof options.socketOptions.connectionTimeout == 'number') {
    finalOptions.connectionTimeout = options.socketOptions.connectionTimeout;
  }

  // Socket timeout
  if(options.socketOptions && typeof options.socketOptions.socketTimeout == 'number') {
    finalOptions.socketTimeout = options.socketOptions.socketTimeout;
  }

  // noDelay
  if(options.socketOptions && typeof options.socketOptions.noDelay == 'boolean') {
    finalOptions.noDelay = options.socketOptions.noDelay;
  }

  if(typeof options.secondaryAcceptableLatencyMS == 'number') {
    finalOptions.acceptableLatency = options.secondaryAcceptableLatencyMS;
  }

  if(options.connectWithNoPrimary == true) {
    finalOptions.secondaryOnlyConnectionAllowed = true;
  }

  // Add the non connection store
  finalOptions.disconnectHandler = store;

  // Translate the options
  if(options.sslCA) finalOptions.ca = options.sslCA;
  if(typeof options.sslValidate == 'boolean') finalOptions.rejectUnauthorized = options.sslValidate;
  if(options.sslKey) finalOptions.key = options.sslKey;
  if(options.sslCert) finalOptions.cert = options.sslCert;
  if(options.sslPass) finalOptions.passphrase = options.sslPass;

  // Create the ReplSet
  var replset = new CReplSet(seedlist, finalOptions)
  // Server capabilities
  var sCapabilities = null;
  // Add auth prbufferMaxEntriesoviders
  replset.addAuthProvider('mongocr', new MongoCR());

  // Listen to reconnect event
  replset.on('reconnect', function() {
    self.emit('reconnect');
    store.execute();
  });
  
  // Ensure the right read Preference object
  var translateReadPreference = function(options) {
    if(typeof options.readPreference == 'string') {
      options.readPreference = new CoreReadPreference(options.readPreference);
    } else if(options.readPreference instanceof ReadPreference) {
      options.readPreference = new CoreReadPreference(options.readPreference.mode
        , options.readPreference.tags);
    }

    return options;  
  }

  // Debug
  if(debug) {
    // Last ismaster
    Object.defineProperty(this, 'replset', {
      enumerable:true, get: function() { return replset; }
    });    
  }

  // Last ismaster
  Object.defineProperty(this, 'isMasterDoc', {
    enumerable:true, get: function() { return replset.lastIsMaster(); }
  });

  // BSON property
  Object.defineProperty(this, 'bson', { 
    enumerable: true, get: function() { 
      return replset.bson; 
    }
  });

  Object.defineProperty(this, 'haInterval', {
    enumerable:true, get: function() { return replset.haInterval; }
  });

  this.parserType = function() {
    return replset.parserType();
  }

  // Connect method
  this.connect = function(db, _options, callback) {
    var self = this;
    if('function' === typeof _options) callback = _options, _options = {};
    if(_options == null) _options = {};
    if(!('function' === typeof callback)) callback = null;
    options = _options;

    // Update bufferMaxEntries
    storeOptions.bufferMaxEntries = db.bufferMaxEntries;

    // Actual handler
    var errorHandler = function(event) {
      return function(err) {
        if(event != 'error') {
          self.emit(event, err);
        }
      }
    }

    // Connect handler
    var connectHandler = function() {
      // Clear out all the current handlers left over
      ["timeout", "error", "close"].forEach(function(e) {
        replset.removeAllListeners(e);
      });

      // Set up listeners
      replset.once('timeout',  errorHandler('timeout'));
      replset.once('error',  errorHandler('error'));
      replset.once('close', errorHandler('close'));

      // relay the event
      var relay = function(event) {
        return function(t, server) {
          self.emit(event, t, server);
        }
      }

      // Replset events relay
      var replsetRelay = function(event) {
        return function(t, server) {
          self.emit(event, t, server.lastIsMaster(), server);
        }
      }

      // Relay ha
      var relayHa = function(t, state) {
        self.emit('ha', t, state);          

        if(t == 'start') {
          self.emit('ha_connect', t, state);          
        } else if(t == 'end') {
          self.emit('ha_ismaster', t, state);          
        }
      }

      // Set up serverConfig listeners
      replset.on('joined', replsetRelay('joined'));
      replset.on('left', relay('left'));
      replset.on('ping', relay('ping'));
      replset.on('ha', relayHa);
      replset.on('fullsetup', relay('fullsetup'));

      // Emit open event
      self.emit('open', null, self);      

      // Return correctly
      try {
        callback(null, self);
      } catch(err) { 
        process.nextTick(function() { throw err; })
      }      
    }

    // Error handler
    var connectErrorHandler = function(event) {
      return function(err) {
        ['timeout', 'error', 'close'].forEach(function(e) {
          replset.removeListener(e, connectErrorHandler);
        });

        replset.removeListener('connect', connectErrorHandler);
        // Destroy the replset
        replset.destroy();

        // Try to callback
        try {
          callback(err);
        } catch(err) { 
          if(!replset.isConnected())
            process.nextTick(function() { throw err; })
        }
      }
    }    

    // Set up listeners
    replset.once('timeout',  connectErrorHandler('timeout'));
    replset.once('error',  connectErrorHandler('error'));
    replset.once('close', connectErrorHandler('close'));
    replset.once('connect', connectHandler);

    // Start connection
    replset.connect(_options);
  }  

  // Server capabilities
  this.capabilities = function() {
    if(sCapabilities) return sCapabilities;
    sCapabilities = new ServerCapabilities(replset.lastIsMaster());
    return sCapabilities;
  }

  // Command
  this.command = function(ns, cmd, options, callback) {
    options = translateReadPreference(options);
    replset.command(ns, cmd, options, callback);
  }

  // Insert
  this.insert = function(ns, ops, options, callback) {
    replset.insert(ns, ops, options, callback);
  }

  // Update
  this.update = function(ns, ops, options, callback) {
    replset.update(ns, ops, options, callback);
  }

  // Remove
  this.remove = function(ns, ops, options, callback) {
    replset.remove(ns, ops, options, callback);
  }

  // IsConnected
  this.isConnected = function() {
    return replset.isConnected();
  }

  this.setBSONParserType = function(type) {
    return replset.setBSONParserType(type);
  }  

  // Insert
  this.cursor = function(ns, cmd, options) {
    options = translateReadPreference(options);
    options.disconnectHandler = store;
    return replset.cursor(ns, cmd, options);
  }

  this.lastIsMaster = function() {
    return replset.lastIsMaster();
  }

  this.close = function(forceClosed) {
    replset.destroy();
    // We need to wash out all stored processes
    if(forceClosed == true) {
      storeOptions.force = forceClosed;
      store.flush();
    }

    var events = ['timeout', 'error', 'close', 'joined', 'left'];
    events.forEach(function(e) {
      self.removeAllListeners(e);
    });        
  }

  this.auth = function() {
    var args = Array.prototype.slice.call(arguments, 0);
    replset.auth.apply(replset, args);
  }

  /**
   * All raw connections
   * @method
   * @return {array}
   */
  this.connections = function() {
    return replset.connections();
  }    
}

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

/**
 * The replset high availability event
 *
 * @event ReplSet#ha
 * @type {function}
 * @param {string} type The stage in the high availability event (start|end)
 * @param {boolean} data.norepeat This is a repeating high availability process or a single execution only
 * @param {number} data.id The id for this high availability request
 * @param {object} data.state An object containing the information about the current replicaset
 */

/**
 * A server member left the replicaset
 *
 * @event ReplSet#left
 * @type {function}
 * @param {string} type The type of member that left (primary|secondary|arbiter)
 * @param {Server} server The server object that left
 */

/**
 * A server member joined the replicaset
 *
 * @event ReplSet#joined
 * @type {function}
 * @param {string} type The type of member that joined (primary|secondary|arbiter)
 * @param {Server} server The server object that joined
 */

/**
 * ReplSet open event, emitted when replicaset can start processing commands.
 *
 * @event ReplSet#open
 * @type {Replset}
 */

/**
 * ReplSet fullsetup event, emitted when all servers in the topology have been connected to.
 *
 * @event ReplSet#fullsetup
 * @type {Replset}
 */

/**
 * ReplSet close event
 *
 * @event ReplSet#close
 * @type {object}
 */

/**
 * ReplSet error event, emitted if there is an error listener.
 *
 * @event ReplSet#error
 * @type {MongoError}
 */

/**
 * ReplSet timeout event
 *
 * @event ReplSet#timeout
 * @type {object}
 */

/**
 * ReplSet parseError event
 *
 * @event ReplSet#parseError
 * @type {object}
 */

/**
 * @ignore
 */
inherits(ReplSet, EventEmitter);

module.exports = ReplSet;
comments powered by Disqus
Documentation generated by JSDoc 3.3.0-alpha9 on Wed Oct 29 2014 13:10:22 GMT+0100 (CET)