Node.js MongoDB Driver API

Source: lib/bulk/ordered.js

  1. "use strict";
  3. var common = require('./common')
  4. 	, utils = require('../utils')
  5.   , toError = require('../utils').toError
  6. 	, handleCallback = require('../utils').handleCallback
  7. 	, shallowClone = utils.shallowClone
  8.   , BulkWriteResult = common.BulkWriteResult
  9.   , ObjectID = require('mongodb-core').BSON.ObjectID
  10.   , Define = require('../metadata')
  11. 	, BSON = require('mongodb-core').BSON
  12.   , Batch = common.Batch
  13.   , mergeBatchResults = common.mergeBatchResults;
  15. var bson = new BSON([BSON.Binary, BSON.Code, BSON.DBRef, BSON.Decimal128,
  16. 	BSON.Double, BSON.Int32, BSON.Long, BSON.Map, BSON.MaxKey, BSON.MinKey,
  17. 	BSON.ObjectId, BSON.BSONRegExp, BSON.Symbol, BSON.Timestamp]);
  19. /**
  20.  * Create a FindOperatorsOrdered instance (INTERNAL TYPE, do not instantiate directly)
  21.  * @class
  22.  * @return {FindOperatorsOrdered} a FindOperatorsOrdered instance.
  23.  */
  24. var FindOperatorsOrdered = function(self) {
  25.   this.s = self.s;
  26. }
  28. /**
  29.  * Add a single update document to the bulk operation
  30.  *
  31.  * @method
  32.  * @param {object} doc update operations
  33.  * @throws {MongoError}
  34.  * @return {OrderedBulkOperation}
  35.  */
  36. FindOperatorsOrdered.prototype.update = function(updateDocument) {
  37.   // Perform upsert
  38.   var upsert = typeof this.s.currentOp.upsert == 'boolean' ? this.s.currentOp.upsert : false;
  40.   // Establish the update command
  41.   var document = {
  42.       q: this.s.currentOp.selector
  43.     , u: updateDocument
  44.     , multi: true
  45.     , upsert: upsert
  46.   }
  48.   // Clear out current Op
  49.   this.s.currentOp = null;
  50.   // Add the update document to the list
  51.   return addToOperationsList(this, common.UPDATE, document);
  52. }
  54. /**
  55.  * Add a single update one document to the bulk operation
  56.  *
  57.  * @method
  58.  * @param {object} doc update operations
  59.  * @throws {MongoError}
  60.  * @return {OrderedBulkOperation}
  61.  */
  62. FindOperatorsOrdered.prototype.updateOne = function(updateDocument) {
  63.   // Perform upsert
  64.   var upsert = typeof this.s.currentOp.upsert == 'boolean' ? this.s.currentOp.upsert : false;
  66.   // Establish the update command
  67.   var document = {
  68.       q: this.s.currentOp.selector
  69.     , u: updateDocument
  70.     , multi: false
  71.     , upsert: upsert
  72.   }
  74.   // Clear out current Op
  75.   this.s.currentOp = null;
  76.   // Add the update document to the list
  77.   return addToOperationsList(this, common.UPDATE, document);
  78. }
  80. /**
  81.  * Add a replace one operation to the bulk operation
  82.  *
  83.  * @method
  84.  * @param {object} doc the new document to replace the existing one with
  85.  * @throws {MongoError}
  86.  * @return {OrderedBulkOperation}
  87.  */
  88. FindOperatorsOrdered.prototype.replaceOne = function(updateDocument) {
  89.   this.updateOne(updateDocument);
  90. }
  92. /**
  93.  * Upsert modifier for update bulk operation
  94.  *
  95.  * @method
  96.  * @throws {MongoError}
  97.  * @return {FindOperatorsOrdered}
  98.  */
  99. FindOperatorsOrdered.prototype.upsert = function() {
  100.   this.s.currentOp.upsert = true;
  101.   return this;
  102. }
  104. /**
  105.  * Add a remove one operation to the bulk operation
  106.  *
  107.  * @method
  108.  * @throws {MongoError}
  109.  * @return {OrderedBulkOperation}
  110.  */
  111. FindOperatorsOrdered.prototype.deleteOne = function() {
  112.   // Establish the update command
  113.   var document = {
  114.       q: this.s.currentOp.selector
  115.     , limit: 1
  116.   }
  118.   // Clear out current Op
  119.   this.s.currentOp = null;
  120.   // Add the remove document to the list
  121.   return addToOperationsList(this, common.REMOVE, document);
  122. }
  124. // Backward compatibility
  125. FindOperatorsOrdered.prototype.removeOne = FindOperatorsOrdered.prototype.deleteOne;
  127. /**
  128.  * Add a remove operation to the bulk operation
  129.  *
  130.  * @method
  131.  * @throws {MongoError}
  132.  * @return {OrderedBulkOperation}
  133.  */
  134. FindOperatorsOrdered.prototype.delete = function() {
  135.   // Establish the update command
  136.   var document = {
  137.       q: this.s.currentOp.selector
  138.     , limit: 0
  139.   }
  141.   // Clear out current Op
  142.   this.s.currentOp = null;
  143.   // Add the remove document to the list
  144.   return addToOperationsList(this, common.REMOVE, document);
  145. }
  147. // Backward compatibility
  148. FindOperatorsOrdered.prototype.remove = FindOperatorsOrdered.prototype.delete;
  150. // Add to internal list of documents
  151. var addToOperationsList = function(_self, docType, document) {
  152.   // Get the bsonSize
  153.   var bsonSize = bson.calculateObjectSize(document, {
  154. 		checkKeys: false,
  155. 	});
  157.   // Throw error if the doc is bigger than the max BSON size
  158.   if(bsonSize >= _self.s.maxBatchSizeBytes) {
  159. 		throw toError("document is larger than the maximum size " + _self.s.maxBatchSizeBytes);
  160. 	}
  162.   // Create a new batch object if we don't have a current one
  163.   if(_self.s.currentBatch == null) _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  165.   // Check if we need to create a new batch
  166.   if(((_self.s.currentBatchSize + 1) >= _self.s.maxWriteBatchSize)
  167.     || ((_self.s.currentBatchSizeBytes +  _self.s.currentBatchSizeBytes) >= _self.s.maxBatchSizeBytes)
  168.     || (_self.s.currentBatch.batchType != docType)) {
  169.     // Save the batch to the execution stack
  170.     _self.s.batches.push(_self.s.currentBatch);
  172.     // Create a new batch
  173.     _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  175.     // Reset the current size trackers
  176.     _self.s.currentBatchSize = 0;
  177.     _self.s.currentBatchSizeBytes = 0;
  178.   } else {
  179.     // Update current batch size
  180.     _self.s.currentBatchSize = _self.s.currentBatchSize + 1;
  181.     _self.s.currentBatchSizeBytes = _self.s.currentBatchSizeBytes + bsonSize;
  182.   }
  184.   if(docType == common.INSERT) {
  185.     _self.s.bulkResult.insertedIds.push({index: _self.s.currentIndex, _id: document._id});
  186.   }
  188.   // We have an array of documents
  189.   if(Array.isArray(document)) {
  190.     throw toError("operation passed in cannot be an Array");
  191.   } else {
  192.     _self.s.currentBatch.originalIndexes.push(_self.s.currentIndex);
  193.     _self.s.currentBatch.operations.push(document)
  194. 		_self.s.currentBatchSizeBytes = _self.s.currentBatchSizeBytes + bsonSize;
  195.     _self.s.currentIndex = _self.s.currentIndex + 1;
  196.   }
  198.   // Return self
  199.   return _self;
  200. }
  202. /**
  203.  * Create a new OrderedBulkOperation instance (INTERNAL TYPE, do not instantiate directly)
  204.  * @class
  205.  * @property {number} length Get the number of operations in the bulk.
  206.  * @return {OrderedBulkOperation} a OrderedBulkOperation instance.
  207.  */
  208. function OrderedBulkOperation(topology, collection, options) {
  209. 	options = options == null ? {} : options;
  210. 	// TODO Bring from driver information in isMaster
  211. 	var executed = false;
  213. 	// Current item
  214. 	var currentOp = null;
  216. 	// Handle to the bson serializer, used to calculate running sizes
  217. 	var bson = topology.bson;
  219. 	// Namespace for the operation
  220.   var namespace = collection.collectionName;
  222.   // Set max byte size
  223. 	var maxBatchSizeBytes = topology.isMasterDoc && topology.isMasterDoc.maxBsonObjectSize
  224.     ? topology.isMasterDoc.maxBsonObjectSize : (1024*1025*16);
  225. 	var maxWriteBatchSize = topology.isMasterDoc && topology.isMasterDoc.maxWriteBatchSize
  226.     ? topology.isMasterDoc.maxWriteBatchSize : 1000;
  228.   // Get the write concern
  229.   var writeConcern = common.writeConcern(shallowClone(options), collection, options);
  231.   // Get the promiseLibrary
  232.   var promiseLibrary = options.promiseLibrary;
  234.   // No promise library selected fall back
  235.   if(!promiseLibrary) {
  236.     promiseLibrary = typeof global.Promise == 'function' ?
  237.       global.Promise : require('es6-promise').Promise;
  238.   }
  240.   // Final results
  241.   var bulkResult = {
  242.       ok: 1
  243.     , writeErrors: []
  244.     , writeConcernErrors: []
  245.     , insertedIds: []
  246.     , nInserted: 0
  247.     , nUpserted: 0
  248.     , nMatched: 0
  249.     , nModified: 0
  250.     , nRemoved: 0
  251.     , upserted: []
  252.   };
  254.   // Internal state
  255.   this.s = {
  256.     // Final result
  257.       bulkResult: bulkResult
  258.     // Current batch state
  259.     , currentBatch: null
  260.     , currentIndex: 0
  261.     , currentBatchSize: 0
  262.     , currentBatchSizeBytes: 0
  263.     , batches: []
  264.     // Write concern
  265.     , writeConcern: writeConcern
  266.     // Max batch size options
  267.     , maxBatchSizeBytes: maxBatchSizeBytes
  268.     , maxWriteBatchSize: maxWriteBatchSize
  269.     // Namespace
  270.     , namespace: namespace
  271.     // BSON
  272.     , bson: bson
  273.     // Topology
  274.     , topology: topology
  275.     // Options
  276.     , options: options
  277.     // Current operation
  278.     , currentOp: currentOp
  279.     // Executed
  280.     , executed: executed
  281.     // Collection
  282.     , collection: collection
  283.     // Promise Library
  284.     , promiseLibrary: promiseLibrary
  285. 		// Fundamental error
  286. 		, err: null
  287.     // Bypass validation
  288.     , bypassDocumentValidation: typeof options.bypassDocumentValidation == 'boolean' ? options.bypassDocumentValidation : false
  289.   }
  290. }
  292. var define = OrderedBulkOperation.define = new Define('OrderedBulkOperation', OrderedBulkOperation, false);
  294. OrderedBulkOperation.prototype.raw = function(op) {
  295.   var key = Object.keys(op)[0];
  297.   // Set up the force server object id
  298.   var forceServerObjectId = typeof this.s.options.forceServerObjectId == 'boolean'
  299.     ? this.s.options.forceServerObjectId : this.s.collection.s.db.options.forceServerObjectId;
  301.   // Update operations
  302.   if((op.updateOne && op.updateOne.q)
  303.     || (op.updateMany && op.updateMany.q)
  304.     || (op.replaceOne && op.replaceOne.q)) {
  305.     op[key].multi = op.updateOne || op.replaceOne ? false : true;
  306.     return addToOperationsList(this, common.UPDATE, op[key]);
  307.   }
  309.   // Crud spec update format
  310.   if(op.updateOne || op.updateMany || op.replaceOne) {
  311.     var multi = op.updateOne || op.replaceOne ? false : true;
  312.     var operation = {q: op[key].filter, u: op[key].update || op[key].replacement, multi: multi}
  313.     operation.upsert = op[key].upsert ? true: false;
  314. 		if(op.collation) operation.collation = op.collation;
  315.     return addToOperationsList(this, common.UPDATE, operation);
  316.   }
  318.   // Remove operations
  319.   if(op.removeOne || op.removeMany || (op.deleteOne && op.deleteOne.q) || op.deleteMany && op.deleteMany.q) {
  320.     op[key].limit = op.removeOne ? 1 : 0;
  321.     return addToOperationsList(this, common.REMOVE, op[key]);
  322.   }
  324.   // Crud spec delete operations, less efficient
  325.   if(op.deleteOne || op.deleteMany) {
  326.     var limit = op.deleteOne ? 1 : 0;
  327.     operation = {q: op[key].filter, limit: limit}
  328. 		if(op.collation) operation.collation = op.collation;
  329.     return addToOperationsList(this, common.REMOVE, operation);
  330.   }
  332.   // Insert operations
  333.   if(op.insertOne && op.insertOne.document == null) {
  334.     if(forceServerObjectId !== true && op.insertOne._id == null) op.insertOne._id = new ObjectID();
  335.     return addToOperationsList(this, common.INSERT, op.insertOne);
  336.   } else if(op.insertOne && op.insertOne.document) {
  337.     if(forceServerObjectId !== true && op.insertOne.document._id == null) op.insertOne.document._id = new ObjectID();
  338.     return addToOperationsList(this, common.INSERT, op.insertOne.document);
  339.   }
  341.   if(op.insertMany) {
  342.     for(var i = 0; i < op.insertMany.length; i++) {
  343.       if(forceServerObjectId !== true && op.insertMany[i]._id == null) op.insertMany[i]._id = new ObjectID();
  344.       addToOperationsList(this, common.INSERT, op.insertMany[i]);
  345.     }
  347.     return;
  348.   }
  350.   // No valid type of operation
  351.   throw toError("bulkWrite only supports insertOne, insertMany, updateOne, updateMany, removeOne, removeMany, deleteOne, deleteMany");
  352. }
  354. /**
  355.  * Add a single insert document to the bulk operation
  356.  *
  357.  * @param {object} doc the document to insert
  358.  * @throws {MongoError}
  359.  * @return {OrderedBulkOperation}
  360.  */
  361. OrderedBulkOperation.prototype.insert = function(document) {
  362.   if(this.s.collection.s.db.options.forceServerObjectId !== true && document._id == null) document._id = new ObjectID();
  363.   return addToOperationsList(this, common.INSERT, document);
  364. }
  366. /**
  367.  * Initiate a find operation for an update/updateOne/remove/removeOne/replaceOne
  368.  *
  369.  * @method
  370.  * @param {object} selector The selector for the bulk operation.
  371.  * @throws {MongoError}
  372.  * @return {FindOperatorsOrdered}
  373.  */
  374. OrderedBulkOperation.prototype.find = function(selector) {
  375.   if (!selector) {
  376.     throw toError("Bulk find operation must specify a selector");
  377.   }
  379.   // Save a current selector
  380.   this.s.currentOp = {
  381.     selector: selector
  382.   }
  384.   return new FindOperatorsOrdered(this);
  385. }
  387. Object.defineProperty(OrderedBulkOperation.prototype, 'length', {
  388.   enumerable: true,
  389.   get: function() {
  390.     return this.s.currentIndex;
  391.   }
  392. });
  394. //
  395. // Execute next write command in a chain
  396. var executeCommands = function(self, callback) {
  397.   if(self.s.batches.length == 0) {
  398.     return handleCallback(callback, null, new BulkWriteResult(self.s.bulkResult));
  399.   }
  401.   // Ordered execution of the command
  402.   var batch = self.s.batches.shift();
  404.   var resultHandler = function(err, result) {
  405. 		// Error is a driver related error not a bulk op error, terminate
  406. 		if(err && err.driver || err && err.message) {
  407. 			return handleCallback(callback, err);
  408. 		}
  410.     // If we have and error
  411.     if(err) err.ok = 0;
  412.     // Merge the results together
  413.     var mergeResult = mergeBatchResults(true, batch, self.s.bulkResult, err, result);
  414.     if(mergeResult != null) {
  415.       return handleCallback(callback, null, new BulkWriteResult(self.s.bulkResult));
  416.     }
  418.     // If we are ordered and have errors and they are
  419.     // not all replication errors terminate the operation
  420.     if(self.s.bulkResult.writeErrors.length > 0) {
  421.       return handleCallback(callback, toError(self.s.bulkResult.writeErrors[0]), new BulkWriteResult(self.s.bulkResult));
  422.     }
  424.     // Execute the next command in line
  425.     executeCommands(self, callback);
  426.   }
  428.   var finalOptions = {ordered: true}
  429.   if(self.s.writeConcern != null) {
  430.     finalOptions.writeConcern = self.s.writeConcern;
  431.   }
  433. 	// Set an operationIf if provided
  434. 	if(self.operationId) {
  435. 		resultHandler.operationId = self.operationId;
  436. 	}
  438. 	// Serialize functions
  439. 	if(self.s.options.serializeFunctions) {
  440. 		finalOptions.serializeFunctions = true
  441. 	}
  443.   // Serialize functions
  444.   if(self.s.options.ignoreUndefined) {
  445.     finalOptions.ignoreUndefined = true
  446.   }
  448.   // Is the bypassDocumentValidation options specific
  449.   if(self.s.bypassDocumentValidation == true) {
  450.     finalOptions.bypassDocumentValidation = true;
  451.   }
  453.   try {
  454.     if(batch.batchType == common.INSERT) {
  455.       self.s.topology.insert(self.s.collection.namespace, batch.operations, finalOptions, resultHandler);
  456.     } else if(batch.batchType == common.UPDATE) {
  457.       self.s.topology.update(self.s.collection.namespace, batch.operations, finalOptions, resultHandler);
  458.     } else if(batch.batchType == common.REMOVE) {
  459.       self.s.topology.remove(self.s.collection.namespace, batch.operations, finalOptions, resultHandler);
  460.     }
  461.   } catch(err) {
  462.     // Force top level error
  463.     err.ok = 0;
  464.     // Merge top level error and return
  465.     handleCallback(callback, null, mergeBatchResults(false, batch, self.s.bulkResult, err, null));
  466.   }
  467. }
  469. /**
  470.  * The callback format for results
  471.  * @callback OrderedBulkOperation~resultCallback
  472.  * @param {MongoError} error An error instance representing the error during the execution.
  473.  * @param {BulkWriteResult} result The bulk write result.
  474.  */
  476. /**
  477.  * Execute the ordered bulk operation
  478.  *
  479.  * @method
  480.  * @param {object} [options=null] Optional settings.
  481.  * @param {(number|string)} [options.w=null] The write concern.
  482.  * @param {number} [options.wtimeout=null] The write concern timeout.
  483.  * @param {boolean} [options.j=false] Specify a journal write concern.
  484.  * @param {boolean} [options.fsync=false] Specify a file sync write concern.
  485.  * @param {OrderedBulkOperation~resultCallback} [callback] The result callback
  486.  * @throws {MongoError}
  487.  * @return {Promise} returns Promise if no callback passed
  488.  */
  489. OrderedBulkOperation.prototype.execute = function(_writeConcern, callback) {
  490.   var self = this;
  491.   if (this.s.executed) {
  492.     var executedError = toError('batch cannot be re-executed');
  493.     return (typeof callback === 'function') ?
  494.       callback(executedError, null) : this.s.promiseLibrary.reject(executedError);
  495.   }
  497.   if (typeof _writeConcern === 'function') {
  498.     callback = _writeConcern;
  499.   } else if (_writeConcern && typeof _writeConcern === 'object') {
  500.     this.s.writeConcern = _writeConcern;
  501.   }
  503.   // If we have current batch
  504.   if (this.s.currentBatch) this.s.batches.push(this.s.currentBatch)
  506.   // If we have no operations in the bulk raise an error
  507.   if (this.s.batches.length == 0) {
  508.     var emptyBatchError = toError('Invalid Operation, no operations specified');
  509.     return (typeof callback === 'function') ?
  510.       callback(emptyBatchError, null) : this.s.promiseLibrary.reject(emptyBatchError);
  511.   }
  513.   // Execute using callback
  514.   if (typeof callback === 'function') {
  515.     return executeCommands(this, callback);
  516.   }
  518.   // Return a Promise
  519.   return new this.s.promiseLibrary(function(resolve, reject) {
  520.     executeCommands(self, function(err, r) {
  521.       if (err) return reject(err);
  522.       resolve(r);
  523.     });
  524.   });
  525. }
  527. define.classMethod('execute', {callback: true, promise:false});
  529. /**
  530.  * Returns an unordered batch object
  531.  * @ignore
  532.  */
  533. var initializeOrderedBulkOp = function(topology, collection, options) {
  534. 	return new OrderedBulkOperation(topology, collection, options);
  535. }
  537. initializeOrderedBulkOp.OrderedBulkOperation = OrderedBulkOperation;
  538. module.exports = initializeOrderedBulkOp;
  539. module.exports.Bulk = OrderedBulkOperation;