ordered.js 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. 'use strict';
  2. const common = require('./common');
  3. const BulkOperationBase = common.BulkOperationBase;
  4. const utils = require('../utils');
  5. const toError = utils.toError;
  6. const handleCallback = utils.handleCallback;
  7. const BulkWriteResult = common.BulkWriteResult;
  8. const Batch = common.Batch;
  9. const mergeBatchResults = common.mergeBatchResults;
  10. const executeOperation = utils.executeOperation;
  11. const MongoWriteConcernError = require('mongodb-core').MongoWriteConcernError;
  12. const handleMongoWriteConcernError = require('./common').handleMongoWriteConcernError;
  13. const bson = common.bson;
  14. const isPromiseLike = require('../utils').isPromiseLike;
  15. /**
  16. * Add to internal list of Operations
  17. *
  18. * @param {OrderedBulkOperation} bulkOperation
  19. * @param {number} docType number indicating the document type
  20. * @param {object} document
  21. * @return {OrderedBulkOperation}
  22. */
  23. function addToOperationsList(bulkOperation, docType, document) {
  24. // Get the bsonSize
  25. const bsonSize = bson.calculateObjectSize(document, {
  26. checkKeys: false,
  27. // Since we don't know what the user selected for BSON options here,
  28. // err on the safe side, and check the size with ignoreUndefined: false.
  29. ignoreUndefined: false
  30. });
  31. // Throw error if the doc is bigger than the max BSON size
  32. if (bsonSize >= bulkOperation.s.maxBatchSizeBytes)
  33. throw toError('document is larger than the maximum size ' + bulkOperation.s.maxBatchSizeBytes);
  34. // Create a new batch object if we don't have a current one
  35. if (bulkOperation.s.currentBatch == null)
  36. bulkOperation.s.currentBatch = new Batch(docType, bulkOperation.s.currentIndex);
  37. const maxKeySize = bulkOperation.s.maxKeySize;
  38. // Check if we need to create a new batch
  39. if (
  40. bulkOperation.s.currentBatchSize + 1 >= bulkOperation.s.maxWriteBatchSize ||
  41. bulkOperation.s.currentBatchSizeBytes + maxKeySize + bsonSize >=
  42. bulkOperation.s.maxBatchSizeBytes ||
  43. bulkOperation.s.currentBatch.batchType !== docType
  44. ) {
  45. // Save the batch to the execution stack
  46. bulkOperation.s.batches.push(bulkOperation.s.currentBatch);
  47. // Create a new batch
  48. bulkOperation.s.currentBatch = new Batch(docType, bulkOperation.s.currentIndex);
  49. // Reset the current size trackers
  50. bulkOperation.s.currentBatchSize = 0;
  51. bulkOperation.s.currentBatchSizeBytes = 0;
  52. }
  53. if (docType === common.INSERT) {
  54. bulkOperation.s.bulkResult.insertedIds.push({
  55. index: bulkOperation.s.currentIndex,
  56. _id: document._id
  57. });
  58. }
  59. // We have an array of documents
  60. if (Array.isArray(document)) {
  61. throw toError('operation passed in cannot be an Array');
  62. }
  63. bulkOperation.s.currentBatch.originalIndexes.push(bulkOperation.s.currentIndex);
  64. bulkOperation.s.currentBatch.operations.push(document);
  65. bulkOperation.s.currentBatchSize += 1;
  66. bulkOperation.s.currentBatchSizeBytes += maxKeySize + bsonSize;
  67. bulkOperation.s.currentIndex += 1;
  68. // Return bulkOperation
  69. return bulkOperation;
  70. }
  71. /**
  72. * Create a new OrderedBulkOperation instance (INTERNAL TYPE, do not instantiate directly)
  73. * @class
  74. * @property {number} length Get the number of operations in the bulk.
  75. * @return {OrderedBulkOperation} a OrderedBulkOperation instance.
  76. */
  77. class OrderedBulkOperation extends BulkOperationBase {
  78. constructor(topology, collection, options) {
  79. options = options || {};
  80. options = Object.assign(options, { addToOperationsList });
  81. super(topology, collection, options, true);
  82. }
  83. /**
  84. * The callback format for results
  85. * @callback OrderedBulkOperation~resultCallback
  86. * @param {MongoError} error An error instance representing the error during the execution.
  87. * @param {BulkWriteResult} result The bulk write result.
  88. */
  89. /**
  90. * Execute the ordered bulk operation
  91. *
  92. * @method
  93. * @param {object} [options] Optional settings.
  94. * @param {(number|string)} [options.w] The write concern.
  95. * @param {number} [options.wtimeout] The write concern timeout.
  96. * @param {boolean} [options.j=false] Specify a journal write concern.
  97. * @param {boolean} [options.fsync=false] Specify a file sync write concern.
  98. * @param {OrderedBulkOperation~resultCallback} [callback] The result callback
  99. * @throws {MongoError}
  100. * @return {Promise} returns Promise if no callback passed
  101. */
  102. execute(_writeConcern, options, callback) {
  103. const ret = this.bulkExecute(_writeConcern, options, callback);
  104. if (!ret || isPromiseLike(ret)) {
  105. return ret;
  106. }
  107. options = ret.options;
  108. callback = ret.callback;
  109. return executeOperation(this.s.topology, executeCommands, [this, options, callback]);
  110. }
  111. }
  112. /**
  113. * Execute next write command in a chain
  114. *
  115. * @param {OrderedBulkOperation} bulkOperation
  116. * @param {object} options
  117. * @param {function} callback
  118. */
  119. function executeCommands(bulkOperation, options, callback) {
  120. if (bulkOperation.s.batches.length === 0) {
  121. return handleCallback(callback, null, new BulkWriteResult(bulkOperation.s.bulkResult));
  122. }
  123. // Ordered execution of the command
  124. const batch = bulkOperation.s.batches.shift();
  125. function resultHandler(err, result) {
  126. // Error is a driver related error not a bulk op error, terminate
  127. if (((err && err.driver) || (err && err.message)) && !(err instanceof MongoWriteConcernError)) {
  128. return handleCallback(callback, err);
  129. }
  130. // If we have and error
  131. if (err) err.ok = 0;
  132. if (err instanceof MongoWriteConcernError) {
  133. return handleMongoWriteConcernError(batch, bulkOperation.s.bulkResult, true, err, callback);
  134. }
  135. // Merge the results together
  136. const writeResult = new BulkWriteResult(bulkOperation.s.bulkResult);
  137. const mergeResult = mergeBatchResults(true, batch, bulkOperation.s.bulkResult, err, result);
  138. if (mergeResult != null) {
  139. return handleCallback(callback, null, writeResult);
  140. }
  141. if (bulkOperation.handleWriteError(callback, writeResult)) return;
  142. // Execute the next command in line
  143. executeCommands(bulkOperation, options, callback);
  144. }
  145. bulkOperation.finalOptionsHandler({ options, batch, resultHandler }, callback);
  146. }
  147. /**
  148. * Returns an unordered batch object
  149. * @ignore
  150. */
  151. function initializeOrderedBulkOp(topology, collection, options) {
  152. return new OrderedBulkOperation(topology, collection, options);
  153. }
  154. initializeOrderedBulkOp.OrderedBulkOperation = OrderedBulkOperation;
  155. module.exports = initializeOrderedBulkOp;
  156. module.exports.Bulk = OrderedBulkOperation;