mongodb - npm Package Compare versions

Comparing version 1.2.13 to 1.2.14

lib/mongodb/connection/connection_pool.js

 var utils = require('./connection_utils'),
   inherits = require('util').inherits,
   net = require('net'),
+  timers = require('timers'),
   EventEmitter = require('events').EventEmitter,
@@ -9,2 +10,6 @@ inherits = require('util').inherits,
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+processor = process.nextTick
+
 var ConnectionPool = exports.ConnectionPool = function(host, port, poolSize, bson, socketOptions) {
@@ -80,2 +85,12 @@ if(typeof host !== 'string') {
 
+ConnectionPool.prototype.setMaxMessageSizeBytes = function(maxMessageSizeBytes) {
+  if(maxMessageSizeBytes == null){
+    maxMessageSizeBytes = Connection.DEFAULT_MAX_MESSAGE_SIZE;
+  }
+
+  for(var i = 0; i < this.openConnections.length; i++) {
+    this.openConnections[i].maxMessageSizeBytes = maxMessageSizeBytes;
+  }
+}
+
 // Start a function
@@ -109,3 +124,3 @@ var _connect = function(_self) {
         } else {
-          process.nextTick(function() {
+          processor(function() {
             // If we are still connecting (no close events fired in between start another connection)
@@ -112,0 +127,0 @@ if(_self._poolState == 'connecting') {

lib/mongodb/connection/connection.js

@@ -27,2 +27,3 @@ var utils = require('./connection_utils'),
   this.maxBsonSize = socketOptions.maxBsonSize ? socketOptions.maxBsonSize : Connection.DEFAULT_MAX_BSON_SIZE;
+  this.maxMessageSizeBytes = socketOptions.maxMessageSizeBytes ? socketOptions.maxMessageSizeBytes : Connection.DEFAULT_MAX_MESSAGE_SIZE;
   // Contains the current message bytes
@@ -46,2 +47,4 @@ this.buffer = null;
 Connection.DEFAULT_MAX_BSON_SIZE = 1024 * 1024 * 4;
+// Set default to max bson to avoid overflow or bad guesses
+Connection.DEFAULT_MAX_MESSAGE_SIZE = Connection.DEFAULT_MAX_BSON_SIZE;
 
@@ -96,6 +99,2 @@ // Inherit event emitter so we can emit stuff wohoo
       tls_options.key = this.socketOptions.sslKey;
-      // Allow for a combined cert/key pem file being passed in as cert parameter
-      // if(tls_options.key == null) {
-      //   tls_options.key = this.socketOptions.ssl_cert;
-      // }
     }
@@ -193,4 +192,10 @@
         var binaryCommand = command[i].toBinary()
+        
         if(!this.socketOptions['disableDriverBSONSizeCheck'] && binaryCommand.length > this.maxBsonSize) 
-          return callback(new Error("Document exceeds maximal allowed bson size of " + this.maxBsonSize + " bytes"));
+          return callback(new Error("Document exceeds maximum allowed bson size of " + this.maxBsonSize + " bytes"));
+        
+        if(this.socketOptions['disableDriverBSONSizeCheck'] && binaryCommand.length > this.maxMessageSizeBytes) {
+          return callback(new Error("Command exceeds maximum message size of " + this.maxMessageSizeBytes + " bytes"));
+        }
+
         if(this.logger != null && this.logger.doDebug) 
@@ -203,4 +208,9 @@ this.logger.debug("writing command to mongodb", {binary: binaryCommand, json: command[i]});
       var binaryCommand = command.toBinary()
+
       if(!this.socketOptions['disableDriverBSONSizeCheck'] && binaryCommand.length > this.maxBsonSize) 
-        return callback(new Error("Document exceeds maximal allowed bson size of " + this.maxBsonSize + " bytes"));
+        return callback(new Error("Document exceeds maximum allowed bson size of " + this.maxBsonSize + " bytes"));
+
+      if(this.socketOptions['disableDriverBSONSizeCheck'] && binaryCommand.length > this.maxMessageSizeBytes) {
+        return callback(new Error("Command exceeds maximum message size of " + this.maxMessageSizeBytes + " bytes"));
+      }
       
@@ -296,3 +306,3 @@ if(this.logger != null && this.logger.doDebug)
           } catch(err) {
-            var errorObject = {err:"socketHandler", trace:err, bin:buffer, parseState:{
+            var errorObject = {err:"socketHandler", trace:err, bin:self.buffer, parseState:{
               sizeOfMessage:self.sizeOfMessage,
@@ -299,0 +309,0 @@ bytesRead:self.bytesRead,

lib/mongodb/connection/repl_set.js

@@ -108,2 +108,4 @@ var Connection = require('./connection').Connection,
   this._readPreference = null;
+  // HA running
+  this._haRunning = false;
   // Do we record server stats or not
@@ -136,2 +138,5 @@ this.recordQueryStats = false;
 
+  // Ensure correct slave set
+  if(this.readSecondary) this.slaveOk = true;
+
   // Strategy for picking a secondary
@@ -242,4 +247,6 @@ this.secondaryAcceptableLatencyMS = this.options['secondaryAcceptableLatencyMS'] == null ? 15 : this.options['secondaryAcceptableLatencyMS'];
   this._readPreference = preference;
-  // Ensure slaveOk is correct for secodnaries read preference and tags
-  if((this._readPreference == ReadPreference.SECONDARY_PREFERRED || this._readPreference == ReadPreference.SECONDARY)
+  // Ensure slaveOk is correct for secondaries read preference and tags
+  if((this._readPreference == ReadPreference.SECONDARY_PREFERRED 
+    || this._readPreference == ReadPreference.SECONDARY
+    || this._readPreference == ReadPreference.NEAREST)
     || (this._readPreference != null && typeof this._readPreference == 'object')) {
@@ -344,2 +351,3 @@ this.slaveOk = true;
   var self = this;
+  self._haRunning = true;
   return check();
@@ -397,3 +405,3 @@
         // Add error handlers
-        self.on("close", function() {
+        self._haServer.on("close", function() {
           if(self._haServer) self._haServer.close();
@@ -403,3 +411,3 @@ self._haServer = null;
 
-        self.on("error", function() {
+        self._haServer.on("error", function() {
           if(self._haServer) self._haServer.close();
@@ -409,3 +417,3 @@ self._haServer = null;
 
-        self.on("timeout", function() {
+        self._haServer.on("timeout", function() {
           if(self._haServer) self._haServer.close();
@@ -419,3 +427,4 @@ self._haServer = null;
           if("disconnected" == self._serverState) {
-            if(_server.close) _server.close();            
+            if(_server && _server.close) _server.close();            
+            self._haRunning = false;
             return;
@@ -425,3 +434,3 @@ }
           if(err) {
-            if(_server.close) _server.close();
+            if(_server && _server.close) _server.close();
             return check();
@@ -445,7 +454,13 @@ }
         if("disconnected" == self._serverState) {
-          if(_server.close) _server.close();            
+          if(_server && _server.close) _server.close();            
+          self._haRunning = false;
+          self._haServer = null;
           return;
         }
         // If error let's set perform another check
-        if(err) return check();
+        if(err) {
+          // Force new server selection
+          self._haServer = null;
+          return check();
+        }
         // Validate the replicaset            
@@ -518,3 +533,3 @@ self._validateReplicaset(res, db.auths, function() {
  */
-function connectTo (hosts, auths, replset, cb) {
+function connectTo(hosts, auths, replset, cb) {
   var pending = hosts.length;
@@ -543,3 +558,3 @@ if (!pending) return cb();
  */
-function connectToHost (host, auths, replset, cb) {
+function connectToHost(host, auths, replset, cb) {
   var server = createServer(host, replset);
@@ -618,3 +633,3 @@
  */
-function createServer (host, replset) {
+function createServer(host, replset) {
   // copy existing socket options to new server
@@ -838,2 +853,3 @@ var socketOptions = {}
       }
+
       addresses[server.host + ":" + server.port] = server;
@@ -857,3 +873,3 @@ // Connect
       } else{
-        if (self.strategyInstance) {
+        if(self.strategyInstance) {
           self.strategyInstance.start();
@@ -864,3 +880,3 @@ }
         self.emit("fullsetup", null, self.db, self);
-        self.emit("open", null, self.db, self);
+        self.emit("open", null, self.db, self);          
       }
@@ -871,2 +887,4 @@ }
 
+var _fullsetup_emitted = false;
+
 /**
@@ -969,3 +987,4 @@ * Interval state object constructor
   }
-  server.connect(parent, opts, _connectHandler(this, candidateServers, server));
+
+  server.connect(parent, opts, _connectHandler(self, candidateServers, server));    
 }
@@ -988,2 +1007,3 @@
   if(!this.haEnabled) return;
+  if(this._haRunning) return;
   this._enableHA();
@@ -990,0 +1010,0 @@ }

lib/mongodb/connection/server.js

@@ -9,4 +9,9 @@ var Connection = require('./connection').Connection,
   utils = require('../utils'),
+  timers = require('timers'),
   inherits = require('util').inherits;
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+processor = process.nextTick
+
 /**
@@ -23,3 +28,2 @@ * Class representing a single MongoDB Server connection
  *  - **sslPass** {Buffer/String, default:null}, String or buffer containing the certificate password (needs to have a mongod server with ssl support, 2.4 or higher)
- *  - **slaveOk** {Boolean, default:false}, legacy option allowing reads from secondary, use **readPrefrence** instead.
  *  - **poolSize** {Number, default:5}, number of connections in the connection pool, set to 5 as default for legacy reasons.
@@ -323,2 +327,3 @@ *  - **socketOptions** {Object, default:null}, an object containing socket options to use (noDelay:(boolean), keepAlive:(number), connectTimeoutMS:(number), socketTimeoutMS:(number))
     server.connectionPool.setMaxBsonSize(reply.documents[0].maxBsonObjectSize);
+    server.connectionPool.setMaxMessageSizeBytes(reply.documents[0].maxMessageSizeBytes);
     // Set server as connected
@@ -526,3 +531,3 @@ server.connected = true;
       // Throw error in next tick
-      process.nextTick(function() {
+      processor(function() {
         throw err;
@@ -529,0 +534,0 @@ })

lib/mongodb/connection/url_parser.js

@@ -175,2 +175,4 @@ var fs = require('fs'),
         break;
+      case 'authSource':
+        dbOptions.authSource = value;
       case 'wtimeoutMS':
@@ -177,0 +179,0 @@ dbOptions.wtimeoutMS = parseInt(value, 10);

lib/mongodb/cursor.js

@@ -7,4 +7,8 @@ var QueryCommand = require('./commands/query_command').QueryCommand,
   CursorStream = require('./cursorstream'),
+  timers = require('timers'),
   utils = require('./utils');
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+
 /**
@@ -254,3 +258,3 @@ * Constructor for a cursor object that handles all the operations on query result
   //FIX: stack overflow (on deep callback) (cred: https://github.com/limp/node-mongodb-native/commit/27da7e4b2af02035847f262b29837a94bbbf6ce2)
-  process.nextTick(function(){
+  processor(function(){
     // Fetch the next object until there is no more objects
@@ -702,10 +706,12 @@ self.nextObject(function(err, item) {
     self.db._executeQueryCommand(getMoreCommand, command_options, function(err, result) {
+      var cbValue;
+
       // Get more done
       self.state = Cursor.OPEN;
 
+      if(err != null) {
+        return callback(utils.toError(err), null);
+      }
+
       try {
-        if(err != null) {
-          return callback(utils.toError(err), null);
-        }
-
         var isDead = 1 === result.responseFlag && result.cursorId.isZero();
@@ -735,6 +741,5 @@
           if(options.noReturn) {
-            return callback(null, true);
-            callback(null, true);  
+            cbValue = true;
           } else {
-            callback(null, self.items.shift());
+            cbValue = self.items.shift();
           }          
@@ -761,2 +766,3 @@ } else if(self.tailable && !isDead && self.awaitdata) {
       }
+      if (cbValue != null) callback(null, cbValue);
     });
@@ -788,32 +794,2 @@
 
- // *  - **skip** {Number} skip number of documents to skip.
- // *  - **limit** {Number}, limit the number of results to return. -1 has a special meaning and is used by Db.eval. A value of 1 will also be treated as if it were -1.
- // *  - **hint**  {Object}, hint force the query to use a specific index.
- // *  - **explain** {Boolean}, explain return the explaination of the query.
- // *  - **slaveOk** {Boolean}, slaveOk, sets the slaveOk flag on the query wire protocol for secondaries.
- // *  - **snapshot** {Boolean}, snapshot Snapshot mode assures no duplicates are returned.
- // *  - **timeout** {Boolean}, timeout allow the query to timeout.
- // *  - **tailable** {Boolean}, tailable allow the cursor to be tailable.
- // *  - **awaitdata** {Boolean}, awaitdata allow the cursor to wait for data, only applicable for tailable cursor.
- // *  - **batchSize** {Number}, batchSize the number of the subset of results to request the database to return for every request. This should initially be greater than 1 otherwise the database will automatically close the cursor. The batch size can be set to 1 with cursorInstance.batchSize after performing the initial query to the database.
- // *  - **raw** {Boolean}, raw return all query documents as raw buffers (default false).
- // *  - **read** {Boolean}, read specify override of read from source (primary/secondary).
- // *  - **returnKey** {Boolean}, returnKey only return the index key.
- // *  - **maxScan** {Number}, maxScan limit the number of items to scan.
- // *  - **min** {Number}, min set index bounds.
- // *  - **max** {Number}, max set index bounds.
- // *  - **showDiskLoc** {Boolean}, showDiskLoc show disk location of results.
- // *  - **comment** {String}, comment you can put a $comment field on a query to make looking in the profiler logs simpler.
- // *  - **numberOfRetries** {Number}, numberOfRetries if using awaidata specifies the number of times to retry on timeout.
- // *  - **dbName** {String}, dbName override the default dbName.
- // *  - **tailableRetryInterval** {Number}, tailableRetryInterval specify the miliseconds between getMores on tailable cursor.
- // *  - **exhaust** {Boolean}, exhaust have the server send all the documents at once as getMore packets.
- // *  - **partial** {Boolean}, partial have the sharded system return a partial result from mongos.
-
- // *  - **sort** {Array | Object}, set to sort the documents coming back from the query. Array of indexes, [['a', 1]] etc.
-
-// function Cursor(db, collection, selector, fields, skip, limit
-// -  , sort, hint, explain, snapshot, timeout, tailable, batchSize, slaveOk, raw, read
-// -  , returnKey, maxScan, min, max, showDiskLoc, comment, awaitdata, numberOfRetries, dbName, tailableRetry
-
   // Create a new cursor and fetch the plan
@@ -932,7 +908,10 @@ var cursor = new Cursor(this.db, this.collection, this.selector, this.fields, {
  *
+ * Options
+ *  - **transform** {Function} function of type function(object) { return transformed }, allows for transformation of data before emitting.
+ *
  * @return {CursorStream} returns a stream object.
  * @api public
  */
-Cursor.prototype.stream = function stream () {
-  return new CursorStream(this);
+Cursor.prototype.stream = function stream(options) {
+  return new CursorStream(this, options);
 }
@@ -939,0 +918,0 @@

lib/mongodb/cursorstream.js

@@ -0,1 +1,6 @@
+var timers = require('timers');
+
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+
 /**
@@ -11,2 +16,5 @@ * Module dependecies.
  *
+ * Options
+ *  - **transform** {Function} function of type function(object) { return transformed }, allows for transformation of data before emitting.
+ *
  * Events
@@ -21,4 +29,5 @@ *  - **data** {function(item) {}} the data event triggers when a document is ready.
  */
-function CursorStream(cursor) {
+function CursorStream(cursor, options) {
   if(!(this instanceof CursorStream)) return new CursorStream(cursor);
+  options = options ? options : {};
 
@@ -31,7 +40,8 @@ Stream.call(this);
   this._destroyed = null;
+  this.options = options;
 
   // give time to hook up events
   var self = this;
-  process.nextTick(function () {
-    self._init();
+  process.nextTick(function() {
+    self._init();      
   });
@@ -73,7 +83,9 @@ }
 CursorStream.prototype._next = function () {
-  if (this.paused || this._destroyed) return;
+  if(this.paused || this._destroyed) return;
 
   var self = this;
   // Get the next object
-  process.nextTick(function() {
+  processor(function() {
+    if(self.paused || self._destroyed) return;
+
     self._cursor.nextObject(function (err, doc) {
@@ -98,3 +110,4 @@ self._onNextObject(err, doc);
   } else if(doc) {
-    this.emit('data', doc);    
+    var data = typeof this.options.transform == 'function' ? this.options.transform(doc) : doc;
+    this.emit('data', data);
     this._next();
@@ -145,3 +158,3 @@ }
 
-  if (err) {
+  if(err) {
     this.emit('error', err);
@@ -148,0 +161,0 @@ }

lib/mongodb/gridfs/gridstore.js

@@ -14,2 +14,3 @@ /**
   fs = require('fs'),
+  timers = require('timers'),
   util = require('util'),
@@ -20,2 +21,6 @@ inherits = util.inherits,
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+processor = process.nextTick
+
 var REFERENCE_BY_FILENAME = 0,
@@ -331,3 +336,3 @@ REFERENCE_BY_ID = 1;
               } else {
-                return process.nextTick(writeChunk);
+                return processor(writeChunk);
               }
@@ -340,3 +345,3 @@ });
       // Process the first write
-      process.nextTick(writeChunk);
+      processor(writeChunk);
     });
@@ -753,3 +758,3 @@ });
     // Update internal position
-    self.position = finalBuffer.length;
+    self.position = self.position + finalBuffer.length;
     // Check if we don't have a file at all
@@ -821,2 +826,6 @@ if(finalLength == 0 && finalBuffer.length == 0) return callback(new Error("File does not exist"), null);
   var targetPosition = 0;
+
+  // console.dir(targetPosition)
+
+  // Calculate the position
   if(seekLocationFinal == exports.GridStore.IO_SEEK_CUR) {
@@ -830,2 +839,3 @@ targetPosition = self.position + finalPosition;
 
+  // Get the chunk
   var newChunkNumber = Math.floor(targetPosition/self.chunkSize);
@@ -832,0 +842,0 @@ if(newChunkNumber != self.currentChunk.chunkNumber) {

lib/mongodb/gridfs/readstream.js

 var Stream = require('stream').Stream,
+  timers = require('timers'),
   util = require('util');
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+processor = process.nextTick
+
 /**
@@ -43,3 +48,3 @@ * ReadStream
   var self = this;
-  process.nextTick(function() {
+  processor(function() {
     self._execute();
@@ -184,3 +189,3 @@ });
   var self = this;
-  process.nextTick(function() {
+  processor(function() {
     self._execute();
@@ -187,0 +192,0 @@ });

lib/mongodb/responses/mongo_reply.js

@@ -1,3 +0,8 @@
-var Long = require('bson').Long;
+var Long = require('bson').Long
+  , timers = require('timers');
 
+// Set processor, setImmediate if 0.10 otherwise nextTick
+var processor = timers.setImmediate ? timers.setImmediate : process.nextTick;
+processor = process.nextTick
+
 /**
@@ -96,3 +101,3 @@ Reply message from mongo db
         if(object_index < _numberReturned) {
-          process.nextTick(processFunction);
+          processor(processFunction);
         } else {
@@ -99,0 +104,0 @@ callback(null);

package.json

 { "name" :            "mongodb"
 , "description" :     "A node.js driver for MongoDB"
 , "keywords" :        ["mongodb", "mongo", "driver", "db"]
-, "version" :         "1.2.13"
+, "version" :         "1.2.14"
 , "author" :          "Christian Amor Kvalheim <[e-mail]>"
@@ -76,2 +76,4 @@ , "contributors" :  [ "Aaron Heckmann",
 	  , "async": "0.1.22"
+    , "integra": "0.0.1"
+    , "optimist": "latest"
   }
@@ -83,5 +85,5 @@ , "config":         { "native" : false }
 , "engines" :       { "node" : ">=0.6.19" }
-, "scripts": { "test" : "make test_pure" }
+, "scripts": { "test" : "make test_functional" }
 , "licenses" :    [ { "type" :  "Apache License, Version 2.0"
                     , "url" :   "http://www.apache.org/licenses/LICENSE-2.0" } ]
 }

No alert changes

Improved metrics

Total package byte prevSize

602639

increased by0.01%

Lines of code

13970

increased by0.4%

Worsened metrics

Dev dependency count

11

increased by22.22%

No dependency changes