ok

1 files changed, 972 insertions, 0 deletions

diff --git a/node_modules/mongodb/lib/mongodb/connection/repl_set_servers.js b/node_modules/mongodb/lib/mongodb/connection/repl_set_servers.js
new file mode 100644
index 0000000..3e0e15a
--- /dev/null
+++ b/node_modules/mongodb/lib/mongodb/connection/repl_set_servers.js
@@ -0,0 +1,972 @@
+var Connection = require('./connection').Connection,
+  DbCommand = require('../commands/db_command').DbCommand,
+  MongoReply = require('../responses/mongo_reply').MongoReply,
+  debug = require('util').debug,
+  EventEmitter = require('events').EventEmitter,
+  inherits = require('util').inherits,
+  inspect = require('util').inspect,
+  Server = require('./server').Server,  
+  PingStrategy = require('./strategies/ping_strategy').PingStrategy,
+  StatisticsStrategy = require('./strategies/statistics_strategy').StatisticsStrategy;
+
+const STATE_STARTING_PHASE_1 = 0;
+const STATE_PRIMARY = 1;
+const STATE_SECONDARY = 2;
+const STATE_RECOVERING = 3;
+const STATE_FATAL_ERROR = 4;
+const STATE_STARTING_PHASE_2 = 5;
+const STATE_UNKNOWN = 6;
+const STATE_ARBITER = 7;
+const STATE_DOWN = 8;
+const STATE_ROLLBACK = 9;
+
+/**
+* ReplSetServers constructor provides master-slave functionality
+*
+* @param serverArr{Array of type Server}
+* @return constructor of ServerCluster
+*
+*/
+var ReplSetServers = exports.ReplSetServers = function(servers, options) {
+  // Set up event emitter
+  EventEmitter.call(this);  
+  // Set up basic
+  if(!(this instanceof ReplSetServers)) return new ReplSetServers(server, options);
+  
+  var self = this;
+  // Contains the master server entry
+  this.options = options == null ? {} : options;
+  this.reconnectWait = this.options["reconnectWait"] != null ? this.options["reconnectWait"] : 1000;
+  this.retries = this.options["retries"] != null ? this.options["retries"] : 30;
+  this.replicaSet = this.options["rs_name"];
+
+  // Are we allowing reads from secondaries ?
+  this.readSecondary = this.options["read_secondary"];
+  this.slaveOk = true;
+  this.closedConnectionCount = 0;
+  this._used = false;
+
+  // Default poolSize for new server instances
+  this.poolSize = this.options.poolSize == null ? 1 : this.options.poolSize;
+
+  // Set up ssl connections
+  this.ssl = this.options.ssl == null ? false : this.options.ssl;
+
+  // Just keeps list of events we allow
+  this.eventHandlers = {error:[], parseError:[], poolReady:[], message:[], close:[], timeout:[]};
+  // Internal state of server connection
+  this._serverState = 'disconnected';
+  // Read preference
+  this._readPreference = null;
+  // Do we record server stats or not
+  this.recordQueryStats = false;
+    
+  // Get the readPreference
+  var readPreference = this.options['readPreference'];  
+  // Read preference setting
+  if(readPreference != null) {
+    if(readPreference != Server.READ_PRIMARY && readPreference != Server.READ_SECONDARY_ONLY
+      && readPreference != Server.READ_SECONDARY) {
+        throw new Error("Illegal readPreference mode specified, " + readPreference);
+    }
+    
+    // Set read Preference
+    this._readPreference = readPreference;
+  } else {
+    this._readPreference = null;        
+  }
+  
+  // Strategy for picking a secondary
+  this.strategy = this.options['strategy'] == null ? 'statistical' : this.options['strategy'];  
+  // Make sure strategy is one of the two allowed
+  if(this.strategy != null && (this.strategy != 'ping' && this.strategy != 'statistical')) throw new Error("Only ping or statistical strategies allowed");  
+  // Let's set up our strategy object for picking secodaries
+  if(this.strategy == 'ping') {
+    // Create a new instance
+    this.strategyInstance = new PingStrategy(this);
+  } else if(this.strategy == 'statistical') {
+    // Set strategy as statistical
+    this.strategyInstance = new StatisticsStrategy(this);
+    // Add enable query information
+    this.enableRecordQueryStats(true);
+  }  
+  
+  // Set default connection pool options
+  this.socketOptions = this.options.socketOptions != null ? this.options.socketOptions : {};  
+
+  // Set up logger if any set
+  this.logger = this.options.logger != null 
+    && (typeof this.options.logger.debug == 'function') 
+    && (typeof this.options.logger.error == 'function') 
+    && (typeof this.options.logger.debug == 'function') 
+      ? this.options.logger : {error:function(message, object) {}, log:function(message, object) {}, debug:function(message, object) {}};
+  
+  // Ensure all the instances are of type server and auto_reconnect is false
+  if(!Array.isArray(servers) || servers.length == 0) {
+    throw Error("The parameter must be an array of servers and contain at least one server");
+  } else if(Array.isArray(servers) || servers.length > 0) {
+    var count = 0;
+    servers.forEach(function(server) {
+      if(server instanceof Server) count = count + 1;
+      // Ensure no server has reconnect on
+      server.options.auto_reconnect = false;
+    });
+
+    if(count < servers.length) {
+      throw Error("All server entries must be of type Server");
+    } else {
+      this.servers = servers;
+    }
+  }  
+  
+  // Auto Reconnect property
+  Object.defineProperty(this, "autoReconnect", { enumerable: true
+    , get: function () {
+        return true;
+      }
+  });
+
+  // Get Read Preference method
+  Object.defineProperty(this, "readPreference", { enumerable: true
+    , get: function () {
+        if(this._readPreference == null && this.readSecondary) {
+          return Server.READ_SECONDARY;
+        } else if(this._readPreference == null && !this.readSecondary) {
+          return Server.READ_PRIMARY;
+        } else {
+          return this._readPreference;
+        }
+      }
+  });  
+
+  // Db Instances
+  Object.defineProperty(this, "dbInstances", {enumerable:true
+    , get: function() {
+      var servers = this.allServerInstances();
+      return servers[0].dbInstances;
+    }
+  })
+
+  // Auto Reconnect property
+  Object.defineProperty(this, "host", { enumerable: true
+    , get: function () {
+        if (this.primary != null) return this.primary.host;
+      }
+  });
+
+  Object.defineProperty(this, "port", { enumerable: true
+    , get: function () {
+        if (this.primary != null) return this.primary.port;
+      }
+  });
+
+  Object.defineProperty(this, "read", { enumerable: true
+    , get: function () {
+        return this.secondaries.length > 0 ? this.secondaries[0] : null;
+      }
+  });
+
+  // Get list of secondaries
+  Object.defineProperty(this, "secondaries", {enumerable: true
+    , get: function() {              
+        var keys = Object.keys(this._state.secondaries);
+        var array = new Array(keys.length);
+        // Convert secondaries to array
+        for(var i = 0; i < keys.length; i++) {
+          array[i] = this._state.secondaries[keys[i]];
+        }
+        return array;
+      }
+  });
+
+  // Get list of all secondaries including passives
+  Object.defineProperty(this, "allSecondaries", {enumerable: true
+    , get: function() {              
+        return this.secondaries.concat(this.passives);
+      }
+  });
+
+  // Get list of arbiters
+  Object.defineProperty(this, "arbiters", {enumerable: true
+    , get: function() {
+        var keys = Object.keys(this._state.arbiters);
+        var array = new Array(keys.length);
+        // Convert arbiters to array
+        for(var i = 0; i < keys.length; i++) {
+          array[i] = this._state.arbiters[keys[i]];
+        }
+        return array;
+      }
+  });
+
+  // Get list of passives
+  Object.defineProperty(this, "passives", {enumerable: true
+    , get: function() {
+        var keys = Object.keys(this._state.passives);
+        var array = new Array(keys.length);
+        // Convert arbiters to array
+        for(var i = 0; i < keys.length; i++) {
+          array[i] = this._state.passives[keys[i]];
+        }
+        return array;
+      }
+  });
+
+  // Master connection property
+  Object.defineProperty(this, "primary", { enumerable: true
+    , get: function () {
+        return this._state != null ? this._state.master : null;
+      }
+  });  
+};
+
+inherits(ReplSetServers, EventEmitter);
+
+// Allow setting the read preference at the replicaset level
+ReplSetServers.prototype.setReadPreference = function(preference) {
+  // Set read preference
+  this._readPreference = preference;
+  // Ensure slaveOk is correct for secodnaries read preference and tags
+  if((this._readPreference == Server.READ_SECONDARY || this._readPreference == Server.READ_SECONDARY_ONLY) 
+    || (this._readPreference != null && typeof this._readPreference == 'object')) {
+    this.slaveOk = true;
+  }
+}
+
+// Return the used state
+ReplSetServers.prototype._isUsed = function() {  
+  return this._used;
+}
+
+ReplSetServers.prototype.setTarget = function(target) {
+  this.target = target;
+};
+
+ReplSetServers.prototype.isConnected = function() {
+  // Return the state of the replicaset server
+  return this.primary != null && this._state.master != null && this._state.master.isConnected();
+}
+
+Server.prototype.isSetMember = function() {
+  return false;
+}
+
+ReplSetServers.prototype.isPrimary = function(config) {
+  return this.readSecondary && this.secondaries.length > 0 ? false : true;
+}
+
+ReplSetServers.prototype.isReadPrimary = ReplSetServers.prototype.isPrimary;
+
+// Clean up dead connections
+var cleanupConnections = ReplSetServers.cleanupConnections = function(connections, addresses, byTags) {
+  // Ensure we don't have entries in our set with dead connections
+  var keys = Object.keys(connections);
+  for(var i = 0; i < keys.length; i++) {
+    var server = connections[keys[i]];
+    // If it's not connected remove it from the list
+    if(!server.isConnected()) {
+      // Remove from connections and addresses
+      delete connections[keys[i]];
+      delete addresses[keys[i]];
+      // Clean up tags if needed
+      if(server.tags != null && typeof server.tags === 'object') {
+        cleanupTags(server, byTags);
+      }
+    }
+  }  
+}
+
+var cleanupTags = ReplSetServers._cleanupTags = function(server, byTags) {
+  var serverTagKeys = Object.keys(server.tags);
+  // Iterate over all server tags and remove any instances for that tag that matches the current
+  // server
+  for(var i = 0; i < serverTagKeys.length; i++) {
+    // Fetch the value for the tag key
+    var value = server.tags[serverTagKeys[i]];
+
+    // If we got an instance of the server
+    if(byTags[serverTagKeys[i]] != null 
+      && byTags[serverTagKeys[i]][value] != null  
+      && Array.isArray(byTags[serverTagKeys[i]][value])) {
+      // List of clean servers
+      var cleanInstances = [];
+      // We got instances for the particular tag set
+      var instances = byTags[serverTagKeys[i]][value];
+      for(var j = 0; j < instances.length; j++) {
+        var serverInstance = instances[j];              
+        // If we did not find an instance add it to the clean instances
+        if((serverInstance.host + ":" + serverInstance.port) !== (server.host + ":" + server.port)) {
+          cleanInstances.push(serverInstance);
+        }
+      }
+      
+      // Update the byTags list
+      byTags[serverTagKeys[i]][value] = cleanInstances;
+    }
+  }
+}
+
+ReplSetServers.prototype.allServerInstances = function() {
+  var self = this;
+  // Close all the servers (concatenate entire list of servers first for ease)
+  var allServers = self._state.master != null ? [self._state.master] : [];
+
+  // Secondary keys
+  var keys = Object.keys(self._state.secondaries);
+  // Add all secondaries
+  for(var i = 0; i < keys.length; i++) {
+    allServers.push(self._state.secondaries[keys[i]]);
+  }
+
+  // Arbiter keys
+  var keys = Object.keys(self._state.arbiters);
+  // Add all arbiters
+  for(var i = 0; i < keys.length; i++) {
+    allServers.push(self._state.arbiters[keys[i]]);
+  }
+
+  // Passive keys
+  var keys = Object.keys(self._state.passives);
+  // Add all arbiters
+  for(var i = 0; i < keys.length; i++) {
+    allServers.push(self._state.passives[keys[i]]);
+  }
+
+  // Return complete list of all servers
+  return allServers;
+}
+
+// Ensure no callback is left hanging when we have an error
+var __executeAllCallbacksWithError = function(dbInstance, error) {
+  var keys = Object.keys(dbInstance._callBackStore._notReplied);
+  // Iterate over all callbacks
+  for(var i = 0; i < keys.length; i++) {
+    // Delete info object
+    delete dbInstance._callBackStore._notReplied[keys[i]];
+    // Emit the error
+    dbInstance._callBackStore.emit(keys[i], error);
+  }
+}
+
+ReplSetServers.prototype.connect = function(parent, options, callback) {
+  var self = this;
+  var dateStamp = new Date().getTime();
+  if('function' === typeof options) callback = options, options = {};  
+  if(options == null) options = {};
+  if(!('function' === typeof callback)) callback = null;
+  
+  // Keep reference to parent
+  this.db = parent;
+  // Set server state to connecting
+  this._serverState = 'connecting';
+  // Reference to the instance
+  var replSetSelf = this;
+  var serverConnections = this.servers;
+  // Ensure parent can do a slave query if it's set
+  parent.slaveOk = this.slaveOk ? this.slaveOk : parent.slaveOk;
+  // Number of total servers that need to initialized (known servers)
+  var numberOfServersLeftToInitialize = serverConnections.length;
+
+  // Clean up state
+  replSetSelf._state = {'master':null, 'secondaries':{}, 'arbiters':{}, 'passives':{}, 'errors':{}, 'addresses':{}, 'byTags':{}, 'setName':null, 'errorMessages':[], 'members':[]};
+
+  // Create a connection handler
+  var connectionHandler = function(instanceServer) {
+    return function(err, result) {
+      // Don't attempt to connect if we are done
+      // if(replSetSelf._serverState === 'disconnected') return;
+      // Remove a server from the list of intialized servers we need to perform
+      numberOfServersLeftToInitialize = numberOfServersLeftToInitialize - 1;
+      // Add enable query information
+      instanceServer.enableRecordQueryStats(replSetSelf.recordQueryStats);
+      
+      if(err == null && result.documents[0].hosts != null) {
+        // Fetch the isMaster command result
+        var document = result.documents[0];
+        // Break out the results
+        var setName = document.setName;
+        var isMaster = document.ismaster;
+        var secondary = document.secondary;
+        var passive = document.passive;
+        var arbiterOnly = document.arbiterOnly;
+        var hosts = Array.isArray(document.hosts) ? document.hosts : [];
+        var arbiters = Array.isArray(document.arbiters) ? document.arbiters : [];
+        var passives = Array.isArray(document.passives) ? document.passives : [];
+        var tags = document.tags ? document.tags : {};
+        var primary = document.primary;
+        var me = document.me;
+                
+        // Only add server to our internal list if it's a master, secondary or arbiter
+        if(isMaster == true || secondary == true || arbiterOnly == true) {
+          // Handle a closed connection
+          var closeHandler = function(err, server) {
+            var closeServers = function() {
+              // Set the state to disconnected
+              parent._state = 'disconnected';
+              // Shut down the replicaset for now and Fire off all the callbacks sitting with no reply
+              if(replSetSelf._serverState == 'connected') {
+                // Close the replicaset
+                replSetSelf.close(function() {
+                  __executeAllCallbacksWithError(parent, err);
+                  // Ensure single callback only
+                  if(callback != null) {
+                    // Single callback only
+                    var internalCallback = callback;
+                    callback = null;
+                    // Return the error
+                    internalCallback(err, null);
+                  } else {
+                    // If the parent has listeners trigger an event
+                    if(parent.listeners("close").length > 0) {
+                      parent.emit("close", err);
+                    }
+                  }
+                });
+              }
+            }            
+            
+            // Check if this is the primary server, then disconnect otherwise keep going
+            if(replSetSelf._state.master != null) {
+              var primaryAddress = replSetSelf._state.master.host + ":" + replSetSelf._state.master.port;
+              var errorServerAddress = server.host + ":" +  server.port;
+            
+              // Only shut down the set if we have a primary server error
+              if(primaryAddress == errorServerAddress) {
+                closeServers();
+              } else {
+                // Remove from the list of servers
+                delete replSetSelf._state.addresses[errorServerAddress];
+                // Locate one of the lists and remove
+                if(replSetSelf._state.secondaries[errorServerAddress] != null) {
+                  delete replSetSelf._state.secondaries[errorServerAddress];
+                } else if(replSetSelf._state.arbiters[errorServerAddress] != null) {
+                  delete replSetSelf._state.arbiters[errorServerAddress];
+                } else if(replSetSelf._state.passives[errorServerAddress] != null) {
+                  delete replSetSelf._state.passives[errorServerAddress];
+                }  
+                
+                // Check if we are reading from Secondary only
+                if(replSetSelf._readPreference == Server.READ_SECONDARY_ONLY && Object.keys(replSetSelf._state.secondaries).length == 0) {
+                  closeServers();
+                }              
+              }
+            } else {
+              closeServers();
+            }
+          }
+
+          // Handle a connection timeout
+          var timeoutHandler = function(err, server) {
+            var closeServers = function() {
+              // Set the state to disconnected
+              parent._state = 'disconnected';
+              // Shut down the replicaset for now and Fire off all the callbacks sitting with no reply
+              if(replSetSelf._serverState == 'connected') {
+                // Close the replicaset
+                replSetSelf.close(function() {
+                  __executeAllCallbacksWithError(parent, err);
+                  // Ensure single callback only
+                  if(callback != null) {
+                    // Single callback only
+                    var internalCallback = callback;
+                    callback = null;
+                    // Return the error
+                    internalCallback(new Error("connection timed out"), null);
+                  } else {
+                    // If the parent has listeners trigger an event
+                    if(parent.listeners("error").length > 0) {
+                      parent.emit("timeout", new Error("connection timed out"));
+                    }
+                  }
+                });
+              }
+            }            
+            
+            // Check if this is the primary server, then disconnect otherwise keep going
+            if(replSetSelf._state.master != null) {
+              var primaryAddress = replSetSelf._state.master.host + ":" + replSetSelf._state.master.port;
+              var errorServerAddress = server.host + ":" + server.port;
+            
+              // Only shut down the set if we have a primary server error
+              if(primaryAddress == errorServerAddress) {
+                closeServers();
+              } else {
+                // Remove from the list of servers
+                delete replSetSelf._state.addresses[errorServerAddress];
+                // Locate one of the lists and remove
+                if(replSetSelf._state.secondaries[errorServerAddress] != null) {
+                  delete replSetSelf._state.secondaries[errorServerAddress];
+                } else if(replSetSelf._state.arbiters[errorServerAddress] != null) {
+                  delete replSetSelf._state.arbiters[errorServerAddress];
+                } else if(replSetSelf._state.passives[errorServerAddress] != null) {
+                  delete replSetSelf._state.passives[errorServerAddress];
+                }  
+                
+                // Check if we are reading from Secondary only
+                if(replSetSelf._readPreference == Server.READ_SECONDARY_ONLY && Object.keys(replSetSelf._state.secondaries).length == 0) {
+                  closeServers();
+                }              
+              }
+            } else {
+              closeServers();
+            }
+          }
+
+          // Handle an error
+          var errorHandler = function(err, server) {
+            var closeServers = function() {
+             // Set the state to disconnected
+              parent._state = 'disconnected';
+              // Shut down the replicaset for now and Fire off all the callbacks sitting with no reply
+              if(replSetSelf._serverState == 'connected') {
+                // Close the replicaset
+                replSetSelf.close(function() {
+                  __executeAllCallbacksWithError(parent, err);
+                  // Ensure single callback only
+                  if(callback != null) {
+                    // Single callback only
+                    var internalCallback = callback;
+                    callback = null;
+                    // Return the error
+                    internalCallback(err, null);
+                  } else {
+                    // If the parent has listeners trigger an event
+                    if(parent.listeners("error").length > 0) {
+                      parent.emit("error", err);
+                    }
+                  }
+                });
+              }
+            }            
+            
+            // Check if this is the primary server, then disconnect otherwise keep going
+            if(replSetSelf._state.master != null) {
+              var primaryAddress = replSetSelf._state.master.host + ":" + replSetSelf._state.master.port;
+              var errorServerAddress = server.host + ":" + server.port;
+            
+              // Only shut down the set if we have a primary server error
+              if(primaryAddress == errorServerAddress) {
+                closeServers();
+              } else {
+                // Remove from the list of servers
+                delete replSetSelf._state.addresses[errorServerAddress];
+                // Locate one of the lists and remove
+                if(replSetSelf._state.secondaries[errorServerAddress] != null) {
+                  delete replSetSelf._state.secondaries[errorServerAddress];
+                } else if(replSetSelf._state.arbiters[errorServerAddress] != null) {
+                  delete replSetSelf._state.arbiters[errorServerAddress];
+                } else if(replSetSelf._state.passives[errorServerAddress] != null) {
+                  delete replSetSelf._state.passives[errorServerAddress];
+                }  
+                
+                // Check if we are reading from Secondary only
+                if(replSetSelf._readPreference == Server.READ_SECONDARY_ONLY && Object.keys(replSetSelf._state.secondaries).length == 0) {
+                  closeServers();
+                }              
+              }
+            } else {
+              closeServers();
+            }
+          }
+          
+          // Ensure we don't have duplicate handlers
+          instanceServer.removeAllListeners("close");
+          instanceServer.removeAllListeners("error");
+          instanceServer.removeAllListeners("timeout");
+
+          // Add error handler to the instance of the server
+          instanceServer.on("close", closeHandler);
+          // Add error handler to the instance of the server
+          instanceServer.on("error", errorHandler);
+          // instanceServer.on("timeout", errorHandler);
+          instanceServer.on("timeout", timeoutHandler);
+          // Add tag info
+          instanceServer.tags = tags;
+
+          // For each tag in tags let's add the instance Server to the list for that tag
+          if(tags != null && typeof tags === 'object') {
+            var tagKeys = Object.keys(tags);
+            // For each tag file in the server add it to byTags
+            for(var i = 0; i < tagKeys.length; i++) {
+              var value = tags[tagKeys[i]];
+              // Check if we have a top level tag object
+              if(replSetSelf._state.byTags[tagKeys[i]] == null) replSetSelf._state.byTags[tagKeys[i]] = {};
+              // For the value check if we have an array of server instances
+              if(!Array.isArray(replSetSelf._state.byTags[tagKeys[i]][value])) replSetSelf._state.byTags[tagKeys[i]][value] = [];
+              // Check that the instance is not already registered there
+              var valueArray = replSetSelf._state.byTags[tagKeys[i]][value];            
+              var found = false;
+
+              // Iterate over all values
+              for(var j = 0; j < valueArray.length; j++) {
+                if(valueArray[j].host == instanceServer.host && valueArray[j].port == instanceServer.port) {
+                  found = true;
+                  break;
+                }
+              }
+
+              // If it was not found push the instance server to the list
+              if(!found) valueArray.push(instanceServer);
+            }
+          }
+
+          // Remove from error list
+          delete replSetSelf._state.errors[me];
+
+          // Add our server to the list of finished servers
+          replSetSelf._state.addresses[me] = instanceServer;
+
+          // Assign the set name
+          if(replSetSelf.replicaSet == null) {
+            replSetSelf._state.setName = setName;          
+          } else if(replSetSelf.replicaSet != setName && replSetSelf._serverState != 'disconnected') {
+            replSetSelf._state.errorMessages.push(new Error("configured mongodb replicaset does not match provided replicaset [" + setName + "] != [" + replSetSelf.replicaSet + "]"));
+            // Set done
+            replSetSelf._serverState = 'disconnected';
+            // ensure no callbacks get called twice
+            var internalCallback = callback;
+            callback = null;
+            // Return error message ignoring rest of calls
+            return internalCallback(replSetSelf._state.errorMessages[0], parent);
+          }
+
+          // Let's add the server to our list of server types
+          if(secondary == true && (passive == false || passive == null)) {
+            replSetSelf._state.secondaries[me] = instanceServer;
+          } else if(arbiterOnly == true) {
+            replSetSelf._state.arbiters[me] = instanceServer;
+          } else if(secondary == true && passive == true) {
+            replSetSelf._state.passives[me] = instanceServer;
+          } else if(isMaster == true) {
+            replSetSelf._state.master = instanceServer;
+          } else if(isMaster == false && primary != null && replSetSelf._state.addresses[primary]) {
+            replSetSelf._state.master = replSetSelf._state.addresses[primary];
+          }
+
+          // Let's go throught all the "possible" servers in the replicaset
+          var candidateServers = hosts.concat(arbiters).concat(passives);        
+
+          // If we have new servers let's add them
+          for(var i = 0; i < candidateServers.length; i++) {
+            // Fetch the server string
+            var candidateServerString = candidateServers[i];        
+            // Add the server if it's not defined
+            if(replSetSelf._state.addresses[candidateServerString] == null) {            
+              // Split the server string
+              var parts = candidateServerString.split(/:/);
+              if(parts.length == 1) {
+                parts = [parts[0], Connection.DEFAULT_PORT];
+              }
+
+              // Default empty socket options object
+              var socketOptions = {};
+              // If a socket option object exists clone it
+              if(replSetSelf.socketOptions != null) {
+                var keys = Object.keys(replSetSelf.socketOptions);
+                for(var i = 0; i < keys.length;i++) socketOptions[keys[i]] = replSetSelf.socketOptions[keys[i]];
+              }
+
+              // Add host information to socket options
+              socketOptions['host'] = parts[0];
+              socketOptions['port'] = parseInt(parts[1]);
+
+              // Create a new server instance
+              var newServer = new Server(parts[0], parseInt(parts[1]), {auto_reconnect:false, 'socketOptions':socketOptions
+                              , logger:replSetSelf.logger, ssl:replSetSelf.ssl, poolSize:replSetSelf.poolSize});
+              // Set the replicaset instance
+              newServer.replicasetInstance = replSetSelf;              
+
+              // Add handlers
+              newServer.on("close", closeHandler);
+              newServer.on("timeout", timeoutHandler);
+              newServer.on("error", errorHandler);
+
+              // Add server to list, ensuring we don't get a cascade of request to the same server
+              replSetSelf._state.addresses[candidateServerString] = newServer;
+
+              // Add a new server to the total number of servers that need to initialized before we are done
+              numberOfServersLeftToInitialize = numberOfServersLeftToInitialize + 1;
+
+              // Let's set up a new server instance
+              newServer.connect(parent, {returnIsMasterResults: true, eventReceiver:newServer}, connectionHandler(newServer));
+            }
+          }          
+        } else {
+          // Remove the instance from out list of servers
+          delete replSetSelf._state.addresses[me];
+        }
+      }
+      
+      // If done finish up
+      if((numberOfServersLeftToInitialize == 0) && replSetSelf._serverState === 'connecting' && replSetSelf._state.errorMessages.length == 0) {
+        // Set db as connected
+        replSetSelf._serverState = 'connected';
+        // If we don't expect a master let's call back, otherwise we need a master before
+        // the connection is successful
+        if(replSetSelf.masterNotNeeded || replSetSelf._state.master != null) {
+          // If we have a read strategy boot it
+          if(replSetSelf.strategyInstance != null) {
+            // Ensure we have a proper replicaset defined
+            replSetSelf.strategyInstance.replicaset = replSetSelf;
+            // Start strategy
+            replSetSelf.strategyInstance.start(function(err) {
+              // ensure no callbacks get called twice
+              var internalCallback = callback;
+              callback = null;
+              // Perform callback
+              internalCallback(null, parent);
+            })
+          } else {
+            // ensure no callbacks get called twice
+            var internalCallback = callback;
+            callback = null;
+            // Perform callback
+            internalCallback(null, parent);
+          }
+        } else if(replSetSelf.readSecondary == true && Object.keys(replSetSelf._state.secondaries).length > 0) {
+          // If we have a read strategy boot it
+          if(replSetSelf.strategyInstance != null) {
+            // Ensure we have a proper replicaset defined
+            replSetSelf.strategyInstance.replicaset = replSetSelf;
+            // Start strategy
+            replSetSelf.strategyInstance.start(function(err) {
+              // ensure no callbacks get called twice
+              var internalCallback = callback;
+              callback = null;
+              // Perform callback
+              internalCallback(null, parent);
+            })
+          } else {
+            // ensure no callbacks get called twice
+            var internalCallback = callback;
+            callback = null;
+            // Perform callback
+            internalCallback(null, parent);
+          }
+        } else if(replSetSelf.readSecondary == true && Object.keys(replSetSelf._state.secondaries).length == 0) {          
+          replSetSelf._serverState = 'disconnected';
+          // ensure no callbacks get called twice
+          var internalCallback = callback;
+          callback = null;
+          // Force close all server instances
+          replSetSelf.close();
+          // Perform callback
+          internalCallback(new Error("no secondary server found"), null);
+        } else if(typeof callback === 'function'){
+          replSetSelf._serverState = 'disconnected';
+          // ensure no callbacks get called twice
+          var internalCallback = callback;
+          callback = null;
+          // Force close all server instances
+          replSetSelf.close();
+          // Perform callback
+          internalCallback(new Error("no primary server found"), null);            
+        }          
+      } else if((numberOfServersLeftToInitialize == 0) && replSetSelf._state.errorMessages.length > 0 && replSetSelf._serverState != 'disconnected') {
+        // Set done
+        replSetSelf._serverState = 'disconnected';
+        // ensure no callbacks get called twice
+        var internalCallback = callback;
+        callback = null;
+        // Force close all server instances
+        replSetSelf.close();
+        // Callback to signal we are done
+        internalCallback(replSetSelf._state.errorMessages[0], null);          
+      }
+    }
+  }
+  
+  // Ensure we have all registered servers in our set
+  for(var i = 0; i < serverConnections.length; i++) {    
+    replSetSelf._state.addresses[serverConnections[i].host + ':' + serverConnections[i].port] = serverConnections[i];
+  }
+
+  // Initialize all the connections
+  for(var i = 0; i < serverConnections.length; i++) {    
+    // Set up the logger for the server connection
+    serverConnections[i].logger = replSetSelf.logger;
+    // Default empty socket options object
+    var socketOptions = {};
+    // If a socket option object exists clone it
+    if(this.socketOptions != null && typeof this.socketOptions === 'object') {
+      var keys = Object.keys(this.socketOptions);
+      for(var j = 0; j < keys.length;j++) socketOptions[keys[j]] = this.socketOptions[keys[j]];
+    }
+    
+    // If ssl is specified
+    if(replSetSelf.ssl) serverConnections[i].ssl = true;
+
+    // Add host information to socket options
+    socketOptions['host'] = serverConnections[i].host;
+    socketOptions['port'] = serverConnections[i].port;
+    
+    // Set the socket options
+    serverConnections[i].socketOptions = socketOptions;
+    // Set the replicaset instance
+    serverConnections[i].replicasetInstance = replSetSelf;
+    // Connect to server
+    serverConnections[i].connect(parent, {returnIsMasterResults: true, eventReceiver:serverConnections[i]}, connectionHandler(serverConnections[i]));
+  }  
+  
+  // Check if we have an error in the inital set of servers and callback with error
+  if(replSetSelf._state.errorMessages.length > 0 && typeof callback === 'function') {
+    // ensure no callbacks get called twice
+    var internalCallback = callback;
+    callback = null;
+    // Perform callback
+    internalCallback(replSetSelf._state.errorMessages[0], null);
+  }
+}
+
+ReplSetServers.prototype.checkoutWriter = function() {
+  // Establish connection
+  var connection = this._state.master != null ? this._state.master.checkoutWriter() : null;  
+  // Return the connection
+  return connection;
+}
+
+ReplSetServers.prototype.checkoutReader = function() {
+  var connection = null;
+  // If we have specified to read from a secondary server grab a random one and read
+  // from it, otherwise just pass the primary connection
+  if((this.readSecondary == true || this._readPreference == Server.READ_SECONDARY || this._readPreference == Server.READ_SECONDARY_ONLY) && Object.keys(this._state.secondaries).length > 0) {
+    // Checkout a secondary server from the passed in set of servers
+    if(this.strategyInstance != null) {
+      connection = this.strategyInstance.checkoutSecondary();
+    } else {
+      // Pick a random key
+      var keys = Object.keys(this._state.secondaries);
+      var key = keys[Math.floor(Math.random() * keys.length)];
+      connection = this._state.secondaries[key].checkoutReader();
+    }
+  } else if(this._readPreference == Server.READ_SECONDARY_ONLY && Object.keys(this._state.secondaries).length == 0) {
+    connection = null;
+  } else if(this._readPreference != null && typeof this._readPreference === 'object') {
+    // Get all tag keys (used to try to find a server that is valid)
+    var keys = Object.keys(this._readPreference);
+    // final instance server
+    var instanceServer = null;
+    // for each key look for an avilable instance
+    for(var i = 0; i < keys.length; i++) {
+      // Grab subkey value
+      var value = this._readPreference[keys[i]];
+
+      // Check if we have any servers for the tag, if we do pick a random one
+      if(this._state.byTags[keys[i]] != null
+        && this._state.byTags[keys[i]][value] != null
+        && Array.isArray(this._state.byTags[keys[i]][value])
+        && this._state.byTags[keys[i]][value].length > 0) {
+        // Let's grab an available server from the list using a random pick
+        var serverInstances = this._state.byTags[keys[i]][value];
+        // Set instance to return
+        instanceServer = serverInstances[Math.floor(Math.random() * serverInstances.length)];
+        break;
+      }
+    }
+
+    // Return the instance of the server
+    connection = instanceServer != null ? instanceServer.checkoutReader() : this.checkoutWriter();
+  } else {
+    connection = this.checkoutWriter();
+  }
+  // Return the connection
+  return connection;
+}
+
+ReplSetServers.prototype.allRawConnections = function() {
+  // Neeed to build a complete list of all raw connections, start with master server
+  var allConnections = [];
+  // Get connection object
+  var allMasterConnections = this._state.master.connectionPool.getAllConnections();
+  // Add all connections to list
+  allConnections = allConnections.concat(allMasterConnections);
+  
+  // If we have read secondary let's add all secondary servers
+  if(this.readSecondary && Object.keys(this._state.secondaries).length > 0) {
+    // Get all the keys
+    var keys = Object.keys(this._state.secondaries);
+    // For each of the secondaries grab the connections
+    for(var i = 0; i < keys.length; i++) {
+      // Get connection object
+      var secondaryPoolConnections = this._state.secondaries[keys[i]].connectionPool.getAllConnections();
+      // Add all connections to list
+      allConnections = allConnections.concat(secondaryPoolConnections);
+    }
+  }
+  
+  // Return all the conections
+  return allConnections;
+}
+
+ReplSetServers.prototype.enableRecordQueryStats = function(enable) {
+  // Set the global enable record query stats
+  this.recordQueryStats = enable;
+  // Ensure all existing servers already have the flag set, even if the 
+  // connections are up already or we have not connected yet
+  if(this._state != null && this._state.addresses != null) {
+    var keys = Object.keys(this._state.addresses);
+    // Iterate over all server instances and set the  enableRecordQueryStats flag
+    for(var i = 0; i < keys.length; i++) {
+      this._state.addresses[keys[i]].enableRecordQueryStats(enable);
+    }
+  } else if(Array.isArray(this.servers)) {
+    for(var i = 0; i < this.servers.length; i++) {
+      this.servers[i].enableRecordQueryStats(enable);
+    }
+  }
+}
+
+ReplSetServers.prototype.disconnect = function(callback) {
+  this.close(callback);
+}
+
+ReplSetServers.prototype.close = function(callback) {
+  var self = this;  
+  // Set server status as disconnected
+  this._serverState = 'disconnected';  
+  // Get all the server instances and close them
+  var allServers = [];
+  // Make sure we have servers
+  if(this._state['addresses'] != null) {
+    var keys = Object.keys(this._state.addresses);
+    for(var i = 0; i < keys.length; i++) {
+      allServers.push(this._state.addresses[keys[i]]);
+    }    
+  }
+  
+  // Let's process all the closing
+  var numberOfServersToClose = allServers.length;
+
+  // Remove all the listeners
+  self.removeAllListeners();
+
+  // Special case where there are no servers
+  if(allServers.length == 0 && typeof callback === 'function') return callback(null, null);
+
+  // Close the servers
+  for(var i = 0; i < allServers.length; i++) {
+    var server = allServers[i];
+    if(server.isConnected()) {
+      // Close each server
+      server.close(function() {
+        numberOfServersToClose = numberOfServersToClose - 1;
+        // Clear out state if we are done
+        if(numberOfServersToClose == 0) {
+          // Clear out state
+          self._state = {'master':null, 'secondaries':{}, 'arbiters':{}, 'passives':{}, 'errors':{}, 'addresses':{}, 'byTags':{}, 'setName':null, 'errorMessages':[], 'members':[]};
+        }
+
+        // If we are finished perform the call back
+        if(numberOfServersToClose == 0 && typeof callback === 'function') {
+          callback(null);          
+        }
+      })      
+    } else {
+      numberOfServersToClose = numberOfServersToClose - 1;      
+      // If we have no more servers perform the callback
+      if(numberOfServersToClose == 0 && typeof callback === 'function') {
+        callback(null);          
+      }
+    }
+  }
+}
\ No newline at end of file