NODE-729 Support sending writeConcern for commands that write

Author: christkv

lib/collection.js

@@ -1410,6 +1410,9 @@ var rename = function(self, newName, opt, callback) {
   var dropTarget = typeof opt.dropTarget == 'boolean' ? opt.dropTarget : false;
   var cmd = {'renameCollection':renameCollection, 'to':toCollection, 'dropTarget':dropTarget};
 
+  // Decorate command with writeConcern if supported
+  decorateWithWriteConcern(cmd, self, opt);
+
   // Execute against admin
   self.s.db.admin().command(cmd, opt, function(err, doc) {
     if(err) return handleCallback(callback, err, null);
@@ -1436,10 +1439,10 @@ define.classMethod('rename', {callback: true, promise:true});
 Collection.prototype.drop = function(options, callback) {
   var self = this;
   if(typeof options == 'function') callback = options, options = {};
-  options = options | {};
+  options = options || {};
 
   // Execute using callback
-  if(typeof callback == 'function') return self.s.db.dropCollection(self.s.name, callback);
+  if(typeof callback == 'function') return self.s.db.dropCollection(self.s.name, options, callback);
   // Return a Promise
   return new this.s.promiseLibrary(function(resolve, reject) {
     self.s.db.dropCollection(self.s.name, options, function(err, r) {
@@ -1645,7 +1648,10 @@ Collection.prototype.dropIndex = function(indexName, options, callback) {
 
 var dropIndex = function(self, indexName, options, callback) {
   // Delete index command
-  var cmd = {'deleteIndexes':self.s.name, 'index':indexName};
+  var cmd = {'dropIndexes':self.s.name, 'index':indexName};
+
+  // Decorate command with writeConcern if supported
+  decorateWithWriteConcern(cmd, self, options);
 
   // Execute command
   self.s.db.command(cmd, options, function(err, result) {
@@ -1663,11 +1669,15 @@ define.classMethod('dropIndex', {callback: true, promise:true});
  * @param {Collection~resultCallback} [callback] The command result callback
  * @return {Promise} returns Promise if no callback passed
  */
-Collection.prototype.dropIndexes = function(callback) {
+Collection.prototype.dropIndexes = function(options, callback) {
   var self = this;
 
+  // Do we have options
+  if(typeof options == 'function') callback = options, options = {};
+  options = options || {};
+
   // Execute using callback
-  if(typeof callback == 'function') return dropIndexes(self, callback);
+  if(typeof callback == 'function') return dropIndexes(self, options, callback);
 
   // Return a Promise
   return new this.s.promiseLibrary(function(resolve, reject) {
@@ -1678,8 +1688,8 @@ Collection.prototype.dropIndexes = function(callback) {
   });
 }
 
-var dropIndexes = function(self, callback) {
-  self.dropIndex('*', function (err, result) {
+var dropIndexes = function(self, options, callback) {
+  self.dropIndex('*', options, function(err, result) {
     if(err) return handleCallback(callback, err, false);
     handleCallback(callback, null, true);
   });
@@ -1726,6 +1736,9 @@ var reIndex = function(self, options, callback) {
   // Reindex
   var cmd = {'reIndex':self.s.name};
 
+  // Decorate command with writeConcern if supported
+  decorateWithWriteConcern(cmd, self, options);
+
   // Execute the command
   self.s.db.command(cmd, options, function(err, result) {
     if(callback == null) return;
@@ -1760,25 +1773,20 @@ Collection.prototype.listIndexes = function(options) {
     throw new MongoError('cannot connect to server');
   }
 
-  // console.log("!!!!!!!!!!!!!! HEY 0")
   // We have a list collections command
   if(this.s.topology.capabilities().hasListIndexesCommand) {
-    // console.log("!!!!!!!!!!!!!! HEY 0:1")
     // Cursor options
     var cursor = options.batchSize ? {batchSize: options.batchSize} : {}
     // Build the command
     var command = { listIndexes: this.s.name, cursor: cursor };
-    // console.log("!!!!!!!!!!!!!! HEY 0:2")
     // Execute the cursor
     var cursor = this.s.topology.cursor(f('%s.$cmd', this.s.dbName), command, options);
     // Do we have a readPreference, apply it
     if(options.readPreference) cursor.setReadPreference(options.readPreference);
-    // console.log("!!!!!!!!!!!!!! HEY 0:3")
     // Return the cursor
     return cursor;
   }
 
-  // console.log("!!!!!!!!!!!!!! HEY 1")
   // Get the namespace
   var ns = f('%s.system.indexes', this.s.dbName);
   // Get the query
@@ -2452,6 +2460,18 @@ var findAndRemove = function(self, query, sort, options, callback) {
 
 define.classMethod('findAndRemove', {callback: true, promise:true});
 
+function decorateWithWriteConcern(command, self, options) {
+  // Do we support write concerns 3.4 and higher
+  if(self.s.topology.capabilities().commandsTakeWriteConcern) {
+    // Get the write concern settings
+    var finalOptions = writeConcern(shallowClone(options), self.s.db, self, options);
+    // Add the write concern to the command
+    if(finalOptions.writeConcern) {
+      command.writeConcern = finalOptions.writeConcern;
+    }
+  }
+}
+
 /**
  * Execute an aggregation framework pipeline against the collection, needs MongoDB >= 2.2
  * @method
@@ -2469,6 +2489,7 @@ define.classMethod('findAndRemove', {callback: true, promise:true});
  */
 Collection.prototype.aggregate = function(pipeline, options, callback) {
   var self = this;
+
   if(Array.isArray(pipeline)) {
     // Set up callback if one is provided
     if(typeof options == 'function') {
@@ -2510,6 +2531,9 @@ Collection.prototype.aggregate = function(pipeline, options, callback) {
   // Build the command
   var command = { aggregate : this.s.name, pipeline : pipeline};
 
+  // Decorate command with writeConcern if supported
+  decorateWithWriteConcern(command, self, options);
+
   // If we have bypassDocumentValidation set
   if(typeof options.bypassDocumentValidation == 'boolean') {
     command.bypassDocumentValidation = options.bypassDocumentValidation;
@@ -3085,7 +3109,10 @@ var mapReduce = function(self, map, reduce, options, callback) {
   // If we have a read preference and inline is not set as output fail hard
   if((options.readPreference != false && options.readPreference != 'primary')
     && options['out'] && (options['out'].inline != 1 && options['out'] != 'inline')) {
+      // Force readPreference to primary
       options.readPreference = 'primary';
+      // Decorate command with writeConcern if supported
+      decorateWithWriteConcern(mapCommandHash, self, options);
   } else if(self.s.readConcern) {
     mapCommandHash.readConcern = self.s.readConcern;
   }

lib/db.js

@@ -479,6 +479,18 @@ Db.prototype.collection = function(name, options, callback) {
 
 define.classMethod('collection', {callback: true, promise:false, returns: [Collection]});
 
+function decorateWithWriteConcern(command, self, options) {
+  // Do we support write concerns 3.4 and higher
+  if(self.s.topology.capabilities().commandsTakeWriteConcern) {
+    // Get the write concern settings
+    var finalOptions = writeConcern(shallowClone(options), self, options);
+    // Add the write concern to the command
+    if(finalOptions.writeConcern) {
+      command.writeConcern = finalOptions.writeConcern;
+    }
+  }
+}
+
 var createCollection = function(self, name, options, callback) {
   // Get the write concern options
   var finalOptions = writeConcern(shallowClone(options), self, options);
@@ -500,6 +512,9 @@ var createCollection = function(self, name, options, callback) {
       // Create collection command
       var cmd = {'create':name};
 
+      // Decorate command with writeConcern if supported
+      decorateWithWriteConcern(cmd, self, options);
+
       // Add all optional parameters
       for(var n in options) {
         if(options[n] != null && typeof options[n] != 'function')
@@ -799,11 +814,14 @@ define.classMethod('renameCollection', {callback: true, promise:true});
 Db.prototype.dropCollection = function(name, options, callback) {
   var self = this;
   if(typeof options == 'function') callback = options, options = {};
-  options = options | {};
+  options = options || {};
 
   // Command to execute
   var cmd = {'drop':name}
 
+  // Decorate with write concern
+  decorateWithWriteConcern(cmd, self, options);
+
   // options
   options = assign({}, this.s.options, {readPreference: ReadPreference.PRIMARY});
 
@@ -843,10 +861,15 @@ define.classMethod('dropCollection', {callback: true, promise:true});
  * @param {Db~resultCallback} [callback] The results callback
  * @return {Promise} returns Promise if no callback passed
  */
-Db.prototype.dropDatabase = function(callback) {
+Db.prototype.dropDatabase = function(options, callback) {
   var self = this;
+  if(typeof options == 'function') callback = options, options = {};
   // Drop database command
   var cmd = {'dropDatabase':1};
+
+  // Decorate with write concern
+  decorateWithWriteConcern(cmd, self, options);
+
   // Ensure primary only
   var options = assign({}, this.s.options, {readPreference: ReadPreference.PRIMARY});
 
@@ -1703,6 +1726,9 @@ var createIndexUsingCreateIndexes = function(self, name, fieldOrSpec, options, c
   // Create command, apply write concern to command
   var cmd = writeConcern({createIndexes: name, indexes: indexes}, self, options);
 
+  // Decorate command with writeConcern if supported
+  decorateWithWriteConcern(cmd, self, options);
+
   // ReadPreference primary
   options.readPreference = ReadPreference.PRIMARY;
 

lib/topology_base.js

@@ -113,6 +113,7 @@ var ServerCapabilities = function(ismaster) {
   var listCollections = false;
   var listIndexes = false;
   var maxNumberOfDocsInBatch = ismaster.maxWriteBatchSize || 1000;
+  var commandsTakeWriteConcern = false;
 
   if(ismaster.minWireVersion >= 0) {
     textSearch = true;
@@ -132,6 +133,10 @@ var ServerCapabilities = function(ismaster) {
     listIndexes = true;
   }
 
+  if(ismaster.maxWireVersion >= 5) {
+    commandsTakeWriteConcern = true;
+  }
+
   // If no min or max wire version set to 0
   if(ismaster.minWireVersion == null) {
     ismaster.minWireVersion = 0;
@@ -151,6 +156,7 @@ var ServerCapabilities = function(ismaster) {
   setup_get_property(this, "minWireVersion", ismaster.minWireVersion);
   setup_get_property(this, "maxWireVersion", ismaster.maxWireVersion);
   setup_get_property(this, "maxNumberOfDocsInBatch", maxNumberOfDocsInBatch);
+  setup_get_property(this, "commandsTakeWriteConcern", commandsTakeWriteConcern);
 }
 
 exports.Store = Store;