mongodb-core - npm Package Compare versions

Comparing version 1.2.28 to 1.2.30

HISTORY.md

@@ -0,1 +1,10 @@
+1.2.30 12-23-2015
+-----------------
+- Pool allocates size + 1 connections when using replicasets, reserving additional pool connection for monitoring exclusively.
+- Fixes bug when all replicaset members are down, that would cause it to fail to reconnect using the originally provided seedlist.
+
+1.2.29 12-17-2015
+-----------------
+- Correctly emit close event when calling destroy on server topology.
+
 1.2.28 12-13-2015
@@ -2,0 +11,0 @@ -----------------

lib/connection/pool.js

@@ -50,3 +50,4 @@ "use strict";
   this.options = options || {};
-  this.size = typeof options.size == 'number' ? options.size : 5;
+  this.size = typeof options.size == 'number' && !isNaN(options.size) ? options.size : 5;
+
   // Message handler
@@ -64,2 +65,4 @@ this.messageHandler = options.messageHandler;
   this.logger = Logger('Pool', options);
+  // If we are monitoring this server we will create an exclusive reserved socket for that
+  this.monitoring = typeof options.monitoring == 'boolean' ? options.monitoring : false;
   // Pool id
@@ -204,7 +207,19 @@ this.id = _id++;
  */
-Pool.prototype.get = function() {
-  // if(this.dead) return null;
-  var connection = this.connections[this.index++];
-  this.index = this.index % this.connections.length;
-  return connection;
+Pool.prototype.get = function(options) {
+  options = options || {};
+
+  // Set the current index
+  this.index = this.index + 1;
+
+  if(this.connections.length == 1) {
+    return this.connections[0];
+  } else if(this.monitoring && options.monitoring) {
+    return this.connections[this.connections.length - 1];
+  } else if(this.monitoring) {
+    this.index = this.index % (this.connections.length - 1);
+    return this.connections[this.index];
+  } else {
+    this.index = this.index % this.connections.length;
+    return this.connections[this.index];
+  }
 }
@@ -211,0 +226,0 @@

lib/topologies/replset_state.js

@@ -234,8 +234,18 @@ "use strict";
  * @method
+ * @param {boolean} [options.includeArbiters] Include Arbiters in returned server list
  * @return {array}
  */
-State.prototype.getAll = function() {
+State.prototype.getAll = function(options) {
+  options = options || {};
   var servers = [];
   if(this.primary) servers.push(this.primary);
-  return servers.concat(this.secondaries);
+  servers = servers.concat(this.secondaries);
+
+  // Include the arbiters
+  if(options.includeArbiters) {
+    servers = servers.concat(this.arbiters);
+  }
+
+  // return ;
+  return servers;
 }
@@ -246,5 +256,7 @@
  * @method
+ * @param {boolean} [options.includeArbiters] Include Arbiters in returned server list
  * @return {array}
  */
-State.prototype.getAllConnections = function() {
+State.prototype.getAllConnections = function(options) {
+  options = options || {};
   var connections = [];
@@ -256,2 +268,9 @@ if(this.primary) connections = connections.concat(this.primary.connections());
 
+  // Include the arbiters
+  if(options.includeArbiters) {
+    this.arbiters.forEach(function(s) {
+      connections = connections.concat(s.connections());
+    })
+  }
+
   return connections;
@@ -258,0 +277,0 @@ }

lib/topologies/replset.js

@@ -97,2 +97,4 @@ "use strict";
   options = options || {};
+  // Clone the options
+  options = cloneOptions(options);
 
@@ -640,2 +642,6 @@ // Validate seedlist
     opts.emitError = true;
+    // Add a reserved connection for monitoring
+    opts.size = opts.size + 1;
+    opts.monitoring = true;
+    // Set up tags if any
     if(self.s.tag) opts.tag = self.s.tag;
@@ -914,5 +920,41 @@ // Share the auth store
 
+var haveAvailableServers = function(state) {
+  if(state.disconnectedServers.length == 0
+    && state.replState.secondaries.length == 0
+    && state.replState.arbiters.length == 0
+    && state.replState.primary == null) return false;
+    return true;
+}
+
 var replicasetInquirer = function(self, state, norepeat) {
   return function() {
     if(state.replState.state == DESTROYED) return
+
+    // We have no connections we need to reseed the disconnected list
+    if(!haveAvailableServers(state)) {
+      // For all entries in the seedlist build a server instance
+      state.disconnectedServers = state.seedlist.map(function(e) {
+        // Clone options
+        var opts = cloneOptions(state.options);
+        // Add host and port
+        opts.host = e.host;
+        opts.port = e.port;
+        opts.reconnect = false;
+        opts.readPreferenceStrategies = state.readPreferenceStrategies;
+        opts.emitError = true;
+        // Add a reserved connection for monitoring
+        opts.size = opts.size + 1;
+        opts.monitoring = true;
+        // Set up tags if any
+        if(state.tag) opts.tag = stage.tag;
+        // Share the auth store
+        opts.authProviders = state.authProviders;
+        // Create a new Server
+        var server = new Server(opts);
+        // Handle the ismaster
+        server.on('ismaster', handleIsmaster(self));
+        return server;
+      });
+    }
+
     // Process already running don't rerun
@@ -969,3 +1011,3 @@ if(state.highAvailabilityProcessRunning) return;
     // We need to query all servers
-    var servers = state.replState.getAll();
+    var servers = state.replState.getAll({includeArbiters:true});
     var serversLeft = servers.length;
@@ -1006,3 +1048,3 @@
         // Execute ismaster
-        server.command('admin.$cmd', {ismaster:true}, function(err, r) {
+        server.command('admin.$cmd', { ismaster:true },  { monitoring:true }, function(err, r) {
           // Clear out the timeoutServer
@@ -1331,2 +1373,5 @@ clearTimeout(timeoutId);
   opts.emitError = true;
+  // Set the size to size + 1 and mark monitoring
+  opts.size = opts.size + 1;
+  opts.monitoring = true;
 
@@ -1333,0 +1378,0 @@ // Do we have an arbiter set the poolSize to 1

lib/topologies/server.js

@@ -272,3 +272,2 @@ "use strict";
     if(state.reconnect) setTimeout(function() {
-      // state.currentReconnectRetry = state.reconnectTries,
       reconnectServer(self, state)
@@ -668,4 +667,7 @@ }, state.reconnectInterval);
   if(self.s.logger.isDebug()) self.s.logger.debug(f('destroy called on server %s', self.name));
+
   // Emit close
-  if(emitClose && self.listeners('close').length > 0) self.emit('close', self);
+  if(emitClose && self.listeners('close').length > 0) {
+    self.emit('close', null, self);
+  }
 
@@ -705,3 +707,2 @@ // Emit destroy event
   var query = new Query(self.s.bson, ns, cmd, queryOptions);
-
   // Set slave OK
@@ -715,3 +716,3 @@ query.slaveOk = slaveOk(options.readPreference);
   // Get a connection (either passed or from the pool)
-  var connection = options.connection || self.s.pool.get();
+  var connection = options.connection || self.s.pool.get(options);
 
@@ -823,3 +824,5 @@ // Double check if we have a valid connection
   // If we have no connection error
-  if(!self.s.pool.isConnected()) return callback(new MongoError(f("no connection available to server %s", self.name)));
+  if(!self.s.pool.isConnected()) {
+    return callback(new MongoError(f("no connection available to server %s", self.name)));
+  }
 
@@ -860,7 +863,8 @@ // Execute on all connections
   // Notify query start to any read Preference strategies
-  if(self.s.readPreferenceStrategies != null)
+  if(self.s.readPreferenceStrategies != null) {
     notifyStrategies(self, self.s, 'startOperation', [self, queries, new Date()]);
+  }
 
   // Get a connection (either passed or from the pool)
-  var connection = options.connection || self.s.pool.get();
+  var connection = options.connection || self.s.pool.get(options);
 
@@ -867,0 +871,0 @@ // Double check if we have a valid connection

package.json

 {
   "name": "mongodb-core",
-  "version": "1.2.28",
+  "version": "1.2.30",
   "description": "Core MongoDB driver functionality, no bells and whistles and meant for integration not end applications",
@@ -5,0 +5,0 @@ "main": "index.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

448283

increased by0.7%

Lines of code

9029

increased by0.83%

No dependency changes