mongodb-js
diff --git a/‎lib/connection/connection.js‎
Lines changed: 3 additions & 20 deletions b/‎lib/connection/connection.js‎
Lines changed: 3 additions & 20 deletions
diff --git a/‎lib/connection/pool.js‎
Lines changed: 99 additions & 37 deletions b/‎lib/connection/pool.js‎
Lines changed: 99 additions & 37 deletions
diff --git a/‎lib/connection/utils.js‎
Lines changed: 0 additions & 15 deletions b/‎lib/connection/utils.js‎
Lines changed: 0 additions & 15 deletions
diff --git a/‎lib/topologies/server.js‎
Lines changed: 14 additions & 0 deletions b/‎lib/topologies/server.js‎
Lines changed: 14 additions & 0 deletions
diff --git a/‎lib/topologies/shared.js‎
Lines changed: 2 additions & 2 deletions b/‎lib/topologies/shared.js‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎lib/wireprotocol/compression.js‎
Lines changed: 64 additions & 0 deletions b/‎lib/wireprotocol/compression.js‎
Lines changed: 64 additions & 0 deletions
diff --git a/‎lib/wireprotocol/shared.js‎
Lines changed: 12 additions & 1 deletion b/‎lib/wireprotocol/shared.js‎
Lines changed: 12 additions & 1 deletion
@@ -7,8 +7,8 @@ var inherits = require('util').inherits
   , crypto = require('crypto')
   , f = require('util').format
   , debugOptions = require('./utils').debugOptions
-  , parseHeader = require('./utils').parseHeader
-  , compressorIDs = require('./utils').compressorIDs
+  , parseHeader = require('../wireprotocol/shared').parseHeader
+  , decompress = require('../wireprotocol/compression').decompress
   , Response = require('./commands').Response
   , MongoError = require('../error')
   , Logger = require('./logger')
@@ -82,7 +82,7 @@ var Connection = function(messageHandler, options) {
   this.host = options.host || 'localhost';
   this.family = typeof options.family == 'number' ? options.family : 4;
   this.keepAlive = typeof options.keepAlive == 'boolean' ? options.keepAlive : true;
-  this.keepAliveInitialDelay = typeof options.keepAliveInitialDelay == 'number' 
+  this.keepAliveInitialDelay = typeof options.keepAliveInitialDelay == 'number'
     ? options.keepAliveInitialDelay : 300000;
   this.noDelay = typeof options.noDelay == 'boolean' ? options.noDelay : true;
   this.connectionTimeout = typeof options.connectionTimeout == 'number'
@@ -221,23 +221,6 @@ var closeHandler = function(self) {
   }
 }
 
-// Decompress a message using the given compressor
-var decompress = function(compressorID, compressedData, callback) {
-  if (compressorID < 0 || compressorID > compressorIDs.length) {
-    throw new Error('Server sent message compressed using an unsupported compressor. (Received compressor ID ' + compressorID + ')');
-  }
-  switch (compressorID) {
-    case compressorIDs.snappy:
-      Snappy.uncompress(compressedData, callback);
-      break;
-    case compressorIDs.zlib:
-      zlib.inflate(compressedData, callback);
-      break;
-    default:
-      callback(null, compressedData);
-  }
-}
-
 // Handle a message once it is recieved
 var emitMessageHandler = function (self, message) {
   var msgHeader = parseHeader(message);
 
@@ -8,7 +8,14 @@ var inherits = require('util').inherits,
   f = require('util').format,
   Query = require('./commands').Query,
   CommandResult = require('./command_result'),
-  assign = require('../utils').assign;
+  assign = require('../utils').assign,
+  Snappy = require('./utils').retrieveSnappy(),
+  zlib = require('zlib'),
+  MESSAGE_HEADER_SIZE = require('../wireprotocol/shared').MESSAGE_HEADER_SIZE,
+  opcodes = require('../wireprotocol/shared').opcodes,
+  compress = require('../wireprotocol/compression').compress,
+  compressorIDs = require('../wireprotocol/compression').compressorIDs,
+  uncompressibleCommands = require('../wireprotocol/compression').uncompressibleCommands;
 
 var MongoCR = require('../auth/mongocr')
   , X509 = require('../auth/x509')
@@ -918,6 +925,56 @@ Pool.prototype.destroy = function(force) {
   checkStatus();
 }
 
+// Prepare the buffer that Pool.prototype.write() uses to send to the server
+var serializeCommands = function(self, commands, result, callback) {
+  // Base case when there are no more commands to serialize
+  if (commands.length === 0) return callback(null, result);
+
+  // Pop off the zeroth command and serialize it
+  var thisCommand = commands.shift();
+  var originalCommandBuffer = thisCommand.toBin();
+
+  // Check whether we and the server have agreed to use a compressor
+  if (self.options.agreedCompressor && !hasUncompressibleCommands(thisCommand)) {
+    // Transform originalCommandBuffer into OP_COMPRESSED
+    var concatenatedOriginalCommandBuffer = Buffer.concat(originalCommandBuffer);
+    var messageToBeCompressed = concatenatedOriginalCommandBuffer.slice(MESSAGE_HEADER_SIZE);
+
+    // Extract information needed for OP_COMPRESSED from the uncompressed message
+    var originalCommandOpCode = concatenatedOriginalCommandBuffer.readInt32LE(12);
+
+    // Compress the message body
+    compress(self, messageToBeCompressed, function(err, compressedMessage) {
+      if (err) return callback(err, null);
+
+      // Create the msgHeader of OP_COMPRESSED
+      var msgHeader = new Buffer(MESSAGE_HEADER_SIZE);
+      msgHeader.writeInt32LE(MESSAGE_HEADER_SIZE + 9 + compressedMessage.length, 0); // messageLength
+      msgHeader.writeInt32LE(thisCommand.requestId, 4); // requestID
+      msgHeader.writeInt32LE(0, 8); // responseTo (zero)
+      msgHeader.writeInt32LE(opcodes.OP_COMPRESSED, 12); // opCode
+
+      // Create the compression details of OP_COMPRESSED
+      var compressionDetails = new Buffer(9);
+      compressionDetails.writeInt32LE(originalCommandOpCode, 0); // originalOpcode
+      compressionDetails.writeInt32LE(messageToBeCompressed.length, 4); // Size of the uncompressed compressedMessage, excluding the MsgHeader
+      compressionDetails.writeUInt8(compressorIDs[self.options.agreedCompressor], 8); // compressorID
+
+      // Push the concatenation of the OP_COMPRESSED message onto results
+      result.push(Buffer.concat([msgHeader, compressionDetails, compressedMessage]));
+
+      // Continue recursing through the commands array
+      serializeCommands(self, commands, result, callback);
+    })
+  } else {
+    // Push the serialization of the command onto results
+    result.push(originalCommandBuffer);
+
+    // Continue recursing through the commands array
+    serializeCommands(self, commands, result, callback);
+  }
+}
+
 /**
  * Write a message to MongoDB
  * @method
@@ -933,6 +990,11 @@ Pool.prototype.write = function(commands, options, cb) {
   // Always have options
   options = options || {};
 
+  // We need to have a callback function unless the message returns no response
+  if(!(typeof cb == 'function') && !options.noResponse) {
+    throw new MongoError('write method must provide a callback');
+  }
+
   // Pool was destroyed error out
   if(this.state == DESTROYED || this.state == DESTROYING) {
     // Callback with an error
@@ -968,25 +1030,6 @@ Pool.prototype.write = function(commands, options, cb) {
     cb: cb, raw: false, promoteLongs: true, promoteValues: true, promoteBuffers: false, fullResult: false
   };
 
-  var buffer = null
-
-  if(Array.isArray(commands)) {
-    buffer = [];
-
-    for(var i = 0; i < commands.length; i++) {
-      buffer.push(commands[i].toBin());
-    }
-
-    // Get the requestId
-    operation.requestId = commands[commands.length - 1].requestId;
-  } else {
-    operation.requestId = commands.requestId;
-    buffer = commands.toBin();
-  }
-
-  // Set the buffers
-  operation.buffer = buffer;
-
   // Set the options for the parsing
   operation.promoteLongs = typeof options.promoteLongs == 'boolean' ? options.promoteLongs : true;
   operation.promoteValues = typeof options.promoteValues == 'boolean' ? options.promoteValues : true;
@@ -997,7 +1040,6 @@ Pool.prototype.write = function(commands, options, cb) {
   operation.command = typeof options.command == 'boolean' ? options.command : false;
   operation.fullResult = typeof options.fullResult == 'boolean' ? options.fullResult : false;
   operation.noResponse = typeof options.noResponse == 'boolean' ? options.noResponse : false;
-  // operation.requestId = options.requestId;
 
   // Optional per operation socketTimeout
   operation.socketTimeout = options.socketTimeout;
@@ -1007,25 +1049,45 @@ Pool.prototype.write = function(commands, options, cb) {
     operation.socketTimeout = options.socketTimeout;
   }
 
-  // We need to have a callback function unless the message returns no response
-  if(!(typeof cb == 'function') && !options.noResponse) {
-    throw new MongoError('write method must provide a callback');
+  // Ensure commands is an array
+  if (!Array.isArray(commands)) {
+    commands = [commands];
   }
 
-  // If we have a monitoring operation schedule as the very first operation
-  // Otherwise add to back of queue
-  if(options.monitoring) {
-    this.queue.unshift(operation);
-  } else {
-    this.queue.push(operation);
-  }
+  // Get the requestId
+  operation.requestId = commands[commands.length - 1].requestId;
 
-  // Attempt to execute the operation
-  if(!self.executing) {
-    process.nextTick(function() {
-      _execute(self)();
-    });
-  }
+  // Prepare the operation buffer
+  serializeCommands(self, commands, [], function(err, serializedCommands) {
+    if (err) throw err;
+
+    // Set the operation's buffer to the serialization of the commands
+    operation.buffer = serializedCommands;
+
+    // If we have a monitoring operation schedule as the very first operation
+    // Otherwise add to back of queue
+    if(options.monitoring) {
+      self.queue.unshift(operation);
+    } else {
+      self.queue.push(operation);
+    }
+
+    // Attempt to execute the operation
+    if(!self.executing) {
+      process.nextTick(function() {
+        _execute(self)();
+      });
+    }
+  })
+
+}
+
+// Return whether a command contains an uncompressible command term
+// Will return true if command contains no uncompressible command terms
+var hasUncompressibleCommands = function(command) {
+  return uncompressibleCommands.some(function(cmd) {
+    return command.query.hasOwnProperty(cmd);
+  });
 }
 
 // Remove connection method
 
@@ -97,25 +97,10 @@ var retrieveSnappy = function() {
   return snappy;
 }
 
-// Parses the header of a wire protocol message
-var parseHeader = function(message) {
-  return {
-    length: message.readInt32LE(0),
-    requestId: message.readInt32LE(4),
-    responseTo: message.readInt32LE(8),
-    opCode: message.readInt32LE(12)
-  }
-}
-
-exports.compressorIDs = {
-  snappy: 1,
-  zlib: 2
-}
 exports.setProperty = setProperty;
 exports.getProperty = getProperty;
 exports.getSingleProperty = getSingleProperty;
 exports.copy = copy;
 exports.debugOptions = debugOptions;
 exports.retrieveBSON = retrieveBSON;
 exports.retrieveSnappy = retrieveSnappy;
-exports.parseHeader = parseHeader;
@@ -265,6 +265,20 @@ var eventHandler = function(self, event) {
           return;
         }
 
+        // Determine whether the server is instructing us to use a compressor
+        if (result.result && result.result.compression) {
+          for (var i = 0; i < self.s.compression.compressors.length; i++) {
+            if (result.result.compression.indexOf(self.s.compression.compressors[i]) > -1) {
+              self.s.pool.options.agreedCompressor = self.s.compression.compressors[i]
+              break;
+            }
+          }
+
+          if (self.s.compression.zlibCompressionLevel) {
+            self.s.pool.options.zlibCompressionLevel = self.s.compression.zlibCompressionLevel;
+          }
+        }
+
         // Ensure no error emitted after initial connect when reconnecting
         self.initalConnect = false;
         // Save the ismaster
 
@@ -64,11 +64,11 @@ function createCompressionInfo(options) {
   // Check that all supplied compressors are valid
   options.compression.compressors.forEach(function(compressor) {
     if (compressor !== 'snappy' && compressor !== 'zlib') {
-      throw new Error('compressors must be at least one of snappy or zlib')
+      throw new Error('compressors must be at least one of snappy or zlib');
     }
   })
 
-  return options.compression.compressors
+  return options.compression.compressors;
 }
 
 function clone(object) {
 
@@ -0,0 +1,64 @@
+var Snappy = require('../connection/utils').retrieveSnappy(),
+    zlib = require('zlib');
+
+var compressorIDs = {
+  snappy: 1,
+  zlib: 2
+}
+
+var uncompressibleCommands = [
+  'ismaster',
+  'saslStart',
+  'saslContinue',
+  'getnonce',
+  'authenticate',
+  'createUser',
+  'updateUser',
+  'copydbSaslStart',
+  'copydbgetnonce',
+  'copydb'
+];
+
+
+// Facilitate compressing a message using an agreed compressor
+var compress = function(self, dataToBeCompressed, callback) {
+  switch (self.options.agreedCompressor) {
+    case 'snappy':
+      Snappy.compress(dataToBeCompressed, callback);
+      break;
+    case 'zlib':
+      // Determine zlibCompressionLevel
+      var zlibOptions = {};
+      if (self.options.zlibCompressionLevel) {
+        zlibOptions.level = self.options.zlibCompressionLevel
+      }
+      zlib.deflate(dataToBeCompressed, zlibOptions, callback);
+      break;
+    default:
+      throw new Error('Attempt to compress message using unknown compressor \"' + self.options.agreedCompressor + '\".');
+  }
+}
+
+// Decompress a message using the given compressor
+var decompress = function(compressorID, compressedData, callback) {
+  if (compressorID < 0 || compressorID > compressorIDs.length) {
+    throw new Error('Server sent message compressed using an unsupported compressor. (Received compressor ID ' + compressorID + ')');
+  }
+  switch (compressorID) {
+    case compressorIDs.snappy:
+      Snappy.uncompress(compressedData, callback);
+      break;
+    case compressorIDs.zlib:
+      zlib.inflate(compressedData, callback);
+      break;
+    default:
+      callback(null, compressedData);
+  }
+}
+
+module.exports = {
+  compressorIDs: compressorIDs,
+  uncompressibleCommands: uncompressibleCommands,
+  compress: compress,
+  decompress: decompress
+}
@@ -37,8 +37,19 @@ var getReadPreference = function(cmd, options) {
   return readPreference;
 }
 
+// Parses the header of a wire protocol message
+var parseHeader = function(message) {
+  return {
+    length: message.readInt32LE(0),
+    requestId: message.readInt32LE(4),
+    responseTo: message.readInt32LE(8),
+    opCode: message.readInt32LE(12)
+  }
+}
+
 module.exports = {
   getReadPreference: getReadPreference,
   MESSAGE_HEADER_SIZE: MESSAGE_HEADER_SIZE,
-  opcodes: opcodes
+  opcodes: opcodes,
+  parseHeader: parseHeader
 }
Original file line number	Diff line number	Diff line change
`@@ -64,11 +64,11 @@ function createCompressionInfo(options) {`
`64`	`64`	`// Check that all supplied compressors are valid`
`65`	`65`	`options.compression.compressors.forEach(function(compressor) {`
`66`	`66`	`if (compressor !== 'snappy' && compressor !== 'zlib') {`
`67`		`- throw new Error('compressors must be at least one of snappy or zlib')`
	`67`	`+ throw new Error('compressors must be at least one of snappy or zlib');`
`68`	`68`	`}`
`69`	`69`	`})`
`70`	`70`
`71`		`- return options.compression.compressors`
	`71`	`+ return options.compression.compressors;`
`72`	`72`	`}`
`73`	`73`
`74`	`74`	`function clone(object) {`