Ohm-Management - Projektarbeit B-ME
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

common.js 30KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157
  1. 'use strict';
  2. const Long = require('mongodb-core').BSON.Long;
  3. const MongoError = require('mongodb-core').MongoError;
  4. const ObjectID = require('mongodb-core').BSON.ObjectID;
  5. const BSON = require('mongodb-core').BSON;
  6. const MongoWriteConcernError = require('mongodb-core').MongoWriteConcernError;
  7. const toError = require('../utils').toError;
  8. const handleCallback = require('../utils').handleCallback;
  9. const applyRetryableWrites = require('../utils').applyRetryableWrites;
  10. const applyWriteConcern = require('../utils').applyWriteConcern;
  11. const executeOperation = require('../utils').executeOperation;
  12. const isPromiseLike = require('../utils').isPromiseLike;
  13. // Error codes
  14. const WRITE_CONCERN_ERROR = 64;
  15. // Insert types
  16. const INSERT = 1;
  17. const UPDATE = 2;
  18. const REMOVE = 3;
  19. const bson = new BSON([
  20. BSON.Binary,
  21. BSON.Code,
  22. BSON.DBRef,
  23. BSON.Decimal128,
  24. BSON.Double,
  25. BSON.Int32,
  26. BSON.Long,
  27. BSON.Map,
  28. BSON.MaxKey,
  29. BSON.MinKey,
  30. BSON.ObjectId,
  31. BSON.BSONRegExp,
  32. BSON.Symbol,
  33. BSON.Timestamp
  34. ]);
  35. /**
  36. * Keeps the state of a unordered batch so we can rewrite the results
  37. * correctly after command execution
  38. * @ignore
  39. */
  40. class Batch {
  41. constructor(batchType, originalZeroIndex) {
  42. this.originalZeroIndex = originalZeroIndex;
  43. this.currentIndex = 0;
  44. this.originalIndexes = [];
  45. this.batchType = batchType;
  46. this.operations = [];
  47. this.size = 0;
  48. this.sizeBytes = 0;
  49. }
  50. }
  51. /**
  52. * Create a new BulkWriteResult instance (INTERNAL TYPE, do not instantiate directly)
  53. *
  54. * @class
  55. * @return {BulkWriteResult} a BulkWriteResult instance
  56. */
  57. class BulkWriteResult {
  58. constructor(bulkResult) {
  59. this.result = bulkResult;
  60. }
  61. /**
  62. * @return {boolean} ok Did bulk operation correctly execute
  63. */
  64. get ok() {
  65. return this.result.ok;
  66. }
  67. /**
  68. * @return {number} nInserted number of inserted documents
  69. */
  70. get nInserted() {
  71. return this.result.nInserted;
  72. }
  73. /**
  74. * @return {number} nUpserted Number of upserted documents
  75. */
  76. get nUpserted() {
  77. return this.result.nUpserted;
  78. }
  79. /**
  80. * @return {number} nMatched Number of matched documents
  81. */
  82. get nMatched() {
  83. return this.result.nMatched;
  84. }
  85. /**
  86. * @return {number} nModified Number of documents updated physically on disk
  87. */
  88. get nModified() {
  89. return this.result.nModified;
  90. }
  91. /**
  92. * @return {number} nRemoved Number of removed documents
  93. */
  94. get nRemoved() {
  95. return this.result.nRemoved;
  96. }
  97. /**
  98. * Return an array of inserted ids
  99. *
  100. * @return {object[]}
  101. */
  102. getInsertedIds() {
  103. return this.result.insertedIds;
  104. }
  105. /**
  106. * Return an array of upserted ids
  107. *
  108. * @return {object[]}
  109. */
  110. getUpsertedIds() {
  111. return this.result.upserted;
  112. }
  113. /**
  114. * Return the upserted id at position x
  115. *
  116. * @param {number} index the number of the upserted id to return, returns undefined if no result for passed in index
  117. * @return {object}
  118. */
  119. getUpsertedIdAt(index) {
  120. return this.result.upserted[index];
  121. }
  122. /**
  123. * Return raw internal result
  124. *
  125. * @return {object}
  126. */
  127. getRawResponse() {
  128. return this.result;
  129. }
  130. /**
  131. * Returns true if the bulk operation contains a write error
  132. *
  133. * @return {boolean}
  134. */
  135. hasWriteErrors() {
  136. return this.result.writeErrors.length > 0;
  137. }
  138. /**
  139. * Returns the number of write errors off the bulk operation
  140. *
  141. * @return {number}
  142. */
  143. getWriteErrorCount() {
  144. return this.result.writeErrors.length;
  145. }
  146. /**
  147. * Returns a specific write error object
  148. *
  149. * @param {number} index of the write error to return, returns null if there is no result for passed in index
  150. * @return {WriteError}
  151. */
  152. getWriteErrorAt(index) {
  153. if (index < this.result.writeErrors.length) {
  154. return this.result.writeErrors[index];
  155. }
  156. return null;
  157. }
  158. /**
  159. * Retrieve all write errors
  160. *
  161. * @return {object[]}
  162. */
  163. getWriteErrors() {
  164. return this.result.writeErrors;
  165. }
  166. /**
  167. * Retrieve lastOp if available
  168. *
  169. * @return {object}
  170. */
  171. getLastOp() {
  172. return this.result.lastOp;
  173. }
  174. /**
  175. * Retrieve the write concern error if any
  176. *
  177. * @return {WriteConcernError}
  178. */
  179. getWriteConcernError() {
  180. if (this.result.writeConcernErrors.length === 0) {
  181. return null;
  182. } else if (this.result.writeConcernErrors.length === 1) {
  183. // Return the error
  184. return this.result.writeConcernErrors[0];
  185. } else {
  186. // Combine the errors
  187. let errmsg = '';
  188. for (let i = 0; i < this.result.writeConcernErrors.length; i++) {
  189. const err = this.result.writeConcernErrors[i];
  190. errmsg = errmsg + err.errmsg;
  191. // TODO: Something better
  192. if (i === 0) errmsg = errmsg + ' and ';
  193. }
  194. return new WriteConcernError({ errmsg: errmsg, code: WRITE_CONCERN_ERROR });
  195. }
  196. }
  197. /**
  198. * @return {BulkWriteResult} a BulkWriteResult instance
  199. */
  200. toJSON() {
  201. return this.result;
  202. }
  203. /**
  204. * @return {string}
  205. */
  206. toString() {
  207. return `BulkWriteResult(${this.toJSON(this.result)})`;
  208. }
  209. /**
  210. * @return {boolean}
  211. */
  212. isOk() {
  213. return this.result.ok === 1;
  214. }
  215. }
  216. /**
  217. * Create a new WriteConcernError instance (INTERNAL TYPE, do not instantiate directly)
  218. *
  219. * @class
  220. * @return {WriteConcernError} a WriteConcernError instance
  221. */
  222. class WriteConcernError {
  223. constructor(err) {
  224. this.err = err;
  225. }
  226. /**
  227. * @return {number} code Write concern error code.
  228. */
  229. get code() {
  230. return this.err.code;
  231. }
  232. /**
  233. * @return {string} errmsg Write concern error message.
  234. */
  235. get errmsg() {
  236. return this.err.errmsg;
  237. }
  238. /**
  239. * @return {object}
  240. */
  241. toJSON() {
  242. return { code: this.err.code, errmsg: this.err.errmsg };
  243. }
  244. /**
  245. * @return {string}
  246. */
  247. toString() {
  248. return `WriteConcernError(${this.err.errmsg})`;
  249. }
  250. }
  251. /**
  252. * Create a new WriteError instance (INTERNAL TYPE, do not instantiate directly)
  253. *
  254. * @class
  255. * @return {WriteConcernError} a WriteConcernError instance
  256. */
  257. class WriteError {
  258. constructor(err) {
  259. this.err = err;
  260. }
  261. /**
  262. * @return {number} code Write concern error code.
  263. */
  264. get code() {
  265. return this.err.code;
  266. }
  267. /**
  268. * @return {number} index Write concern error original bulk operation index.
  269. */
  270. get index() {
  271. return this.err.index;
  272. }
  273. /**
  274. * @return {string} errmsg Write concern error message.
  275. */
  276. get errmsg() {
  277. return this.err.errmsg;
  278. }
  279. /**
  280. * Define access methods
  281. * @return {object}
  282. */
  283. getOperation() {
  284. return this.err.op;
  285. }
  286. /**
  287. * @return {object}
  288. */
  289. toJSON() {
  290. return { code: this.err.code, index: this.err.index, errmsg: this.err.errmsg, op: this.err.op };
  291. }
  292. /**
  293. * @return {string}
  294. */
  295. toString() {
  296. return `WriteError(${JSON.stringify(this.toJSON())})`;
  297. }
  298. }
  299. /**
  300. * Merges results into shared data structure
  301. * @ignore
  302. */
  303. function mergeBatchResults(batch, bulkResult, err, result) {
  304. // If we have an error set the result to be the err object
  305. if (err) {
  306. result = err;
  307. } else if (result && result.result) {
  308. result = result.result;
  309. } else if (result == null) {
  310. return;
  311. }
  312. // Do we have a top level error stop processing and return
  313. if (result.ok === 0 && bulkResult.ok === 1) {
  314. bulkResult.ok = 0;
  315. const writeError = {
  316. index: 0,
  317. code: result.code || 0,
  318. errmsg: result.message,
  319. op: batch.operations[0]
  320. };
  321. bulkResult.writeErrors.push(new WriteError(writeError));
  322. return;
  323. } else if (result.ok === 0 && bulkResult.ok === 0) {
  324. return;
  325. }
  326. // Deal with opTime if available
  327. if (result.opTime || result.lastOp) {
  328. const opTime = result.lastOp || result.opTime;
  329. let lastOpTS = null;
  330. let lastOpT = null;
  331. // We have a time stamp
  332. if (opTime && opTime._bsontype === 'Timestamp') {
  333. if (bulkResult.lastOp == null) {
  334. bulkResult.lastOp = opTime;
  335. } else if (opTime.greaterThan(bulkResult.lastOp)) {
  336. bulkResult.lastOp = opTime;
  337. }
  338. } else {
  339. // Existing TS
  340. if (bulkResult.lastOp) {
  341. lastOpTS =
  342. typeof bulkResult.lastOp.ts === 'number'
  343. ? Long.fromNumber(bulkResult.lastOp.ts)
  344. : bulkResult.lastOp.ts;
  345. lastOpT =
  346. typeof bulkResult.lastOp.t === 'number'
  347. ? Long.fromNumber(bulkResult.lastOp.t)
  348. : bulkResult.lastOp.t;
  349. }
  350. // Current OpTime TS
  351. const opTimeTS = typeof opTime.ts === 'number' ? Long.fromNumber(opTime.ts) : opTime.ts;
  352. const opTimeT = typeof opTime.t === 'number' ? Long.fromNumber(opTime.t) : opTime.t;
  353. // Compare the opTime's
  354. if (bulkResult.lastOp == null) {
  355. bulkResult.lastOp = opTime;
  356. } else if (opTimeTS.greaterThan(lastOpTS)) {
  357. bulkResult.lastOp = opTime;
  358. } else if (opTimeTS.equals(lastOpTS)) {
  359. if (opTimeT.greaterThan(lastOpT)) {
  360. bulkResult.lastOp = opTime;
  361. }
  362. }
  363. }
  364. }
  365. // If we have an insert Batch type
  366. if (batch.batchType === INSERT && result.n) {
  367. bulkResult.nInserted = bulkResult.nInserted + result.n;
  368. }
  369. // If we have an insert Batch type
  370. if (batch.batchType === REMOVE && result.n) {
  371. bulkResult.nRemoved = bulkResult.nRemoved + result.n;
  372. }
  373. let nUpserted = 0;
  374. // We have an array of upserted values, we need to rewrite the indexes
  375. if (Array.isArray(result.upserted)) {
  376. nUpserted = result.upserted.length;
  377. for (let i = 0; i < result.upserted.length; i++) {
  378. bulkResult.upserted.push({
  379. index: result.upserted[i].index + batch.originalZeroIndex,
  380. _id: result.upserted[i]._id
  381. });
  382. }
  383. } else if (result.upserted) {
  384. nUpserted = 1;
  385. bulkResult.upserted.push({
  386. index: batch.originalZeroIndex,
  387. _id: result.upserted
  388. });
  389. }
  390. // If we have an update Batch type
  391. if (batch.batchType === UPDATE && result.n) {
  392. const nModified = result.nModified;
  393. bulkResult.nUpserted = bulkResult.nUpserted + nUpserted;
  394. bulkResult.nMatched = bulkResult.nMatched + (result.n - nUpserted);
  395. if (typeof nModified === 'number') {
  396. bulkResult.nModified = bulkResult.nModified + nModified;
  397. } else {
  398. bulkResult.nModified = null;
  399. }
  400. }
  401. if (Array.isArray(result.writeErrors)) {
  402. for (let i = 0; i < result.writeErrors.length; i++) {
  403. const writeError = {
  404. index: batch.originalZeroIndex + result.writeErrors[i].index,
  405. code: result.writeErrors[i].code,
  406. errmsg: result.writeErrors[i].errmsg,
  407. op: batch.operations[result.writeErrors[i].index]
  408. };
  409. bulkResult.writeErrors.push(new WriteError(writeError));
  410. }
  411. }
  412. if (result.writeConcernError) {
  413. bulkResult.writeConcernErrors.push(new WriteConcernError(result.writeConcernError));
  414. }
  415. }
  416. function executeCommands(bulkOperation, options, callback) {
  417. if (bulkOperation.s.batches.length === 0) {
  418. return handleCallback(callback, null, new BulkWriteResult(bulkOperation.s.bulkResult));
  419. }
  420. const batch = bulkOperation.s.batches.shift();
  421. function resultHandler(err, result) {
  422. // Error is a driver related error not a bulk op error, terminate
  423. if (((err && err.driver) || (err && err.message)) && !(err instanceof MongoWriteConcernError)) {
  424. return handleCallback(callback, err);
  425. }
  426. // If we have and error
  427. if (err) err.ok = 0;
  428. if (err instanceof MongoWriteConcernError) {
  429. return handleMongoWriteConcernError(batch, bulkOperation.s.bulkResult, err, callback);
  430. }
  431. // Merge the results together
  432. const writeResult = new BulkWriteResult(bulkOperation.s.bulkResult);
  433. const mergeResult = mergeBatchResults(batch, bulkOperation.s.bulkResult, err, result);
  434. if (mergeResult != null) {
  435. return handleCallback(callback, null, writeResult);
  436. }
  437. if (bulkOperation.handleWriteError(callback, writeResult)) return;
  438. // Execute the next command in line
  439. executeCommands(bulkOperation, options, callback);
  440. }
  441. bulkOperation.finalOptionsHandler({ options, batch, resultHandler }, callback);
  442. }
  443. /**
  444. * handles write concern error
  445. *
  446. * @param {object} batch
  447. * @param {object} bulkResult
  448. * @param {boolean} ordered
  449. * @param {WriteConcernError} err
  450. * @param {function} callback
  451. */
  452. function handleMongoWriteConcernError(batch, bulkResult, err, callback) {
  453. mergeBatchResults(batch, bulkResult, null, err.result);
  454. const wrappedWriteConcernError = new WriteConcernError({
  455. errmsg: err.result.writeConcernError.errmsg,
  456. code: err.result.writeConcernError.result
  457. });
  458. return handleCallback(
  459. callback,
  460. new BulkWriteError(toError(wrappedWriteConcernError), new BulkWriteResult(bulkResult)),
  461. null
  462. );
  463. }
  464. /**
  465. * Creates a new BulkWriteError
  466. *
  467. * @class
  468. * @param {Error|string|object} message The error message
  469. * @param {BulkWriteResult} result The result of the bulk write operation
  470. * @return {BulkWriteError} A BulkWriteError instance
  471. * @extends {MongoError}
  472. */
  473. class BulkWriteError extends MongoError {
  474. constructor(error, result) {
  475. const message = error.err || error.errmsg || error.errMessage || error;
  476. super(message);
  477. Object.assign(this, error);
  478. this.name = 'BulkWriteError';
  479. this.result = result;
  480. }
  481. }
  482. /**
  483. * Handles the find operators for the bulk operations
  484. * @class
  485. */
  486. class FindOperators {
  487. /**
  488. * @param {OrderedBulkOperation|UnorderedBulkOperation} bulkOperation
  489. */
  490. constructor(bulkOperation) {
  491. this.s = bulkOperation.s;
  492. }
  493. /**
  494. * Add a single update document to the bulk operation
  495. *
  496. * @method
  497. * @param {object} updateDocument update operations
  498. * @throws {MongoError}
  499. * @return {OrderedBulkOperation|UnordedBulkOperation}
  500. */
  501. update(updateDocument) {
  502. // Perform upsert
  503. const upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  504. // Establish the update command
  505. const document = {
  506. q: this.s.currentOp.selector,
  507. u: updateDocument,
  508. multi: true,
  509. upsert: upsert
  510. };
  511. // Clear out current Op
  512. this.s.currentOp = null;
  513. return this.s.options.addToOperationsList(this, UPDATE, document);
  514. }
  515. /**
  516. * Add a single update one document to the bulk operation
  517. *
  518. * @method
  519. * @param {object} updateDocument update operations
  520. * @throws {MongoError}
  521. * @return {OrderedBulkOperation|UnordedBulkOperation}
  522. */
  523. updateOne(updateDocument) {
  524. // Perform upsert
  525. const upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  526. // Establish the update command
  527. const document = {
  528. q: this.s.currentOp.selector,
  529. u: updateDocument,
  530. multi: false,
  531. upsert: upsert
  532. };
  533. // Clear out current Op
  534. this.s.currentOp = null;
  535. return this.s.options.addToOperationsList(this, UPDATE, document);
  536. }
  537. /**
  538. * Add a replace one operation to the bulk operation
  539. *
  540. * @method
  541. * @param {object} updateDocument the new document to replace the existing one with
  542. * @throws {MongoError}
  543. * @return {OrderedBulkOperation|UnorderedBulkOperation}
  544. */
  545. replaceOne(updateDocument) {
  546. this.updateOne(updateDocument);
  547. }
  548. /**
  549. * Upsert modifier for update bulk operation
  550. *
  551. * @method
  552. * @throws {MongoError}
  553. * @return {FindOperators}
  554. */
  555. upsert() {
  556. this.s.currentOp.upsert = true;
  557. return this;
  558. }
  559. /**
  560. * Add a delete one operation to the bulk operation
  561. *
  562. * @method
  563. * @throws {MongoError}
  564. * @return {OrderedBulkOperation|UnordedBulkOperation}
  565. */
  566. deleteOne() {
  567. // Establish the update command
  568. const document = {
  569. q: this.s.currentOp.selector,
  570. limit: 1
  571. };
  572. // Clear out current Op
  573. this.s.currentOp = null;
  574. return this.s.options.addToOperationsList(this, REMOVE, document);
  575. }
  576. /**
  577. * Add a delete operation to the bulk operation
  578. *
  579. * @method
  580. * @throws {MongoError}
  581. * @return {OrderedBulkOperation|UnordedBulkOperation}
  582. */
  583. delete() {
  584. // Establish the update command
  585. const document = {
  586. q: this.s.currentOp.selector,
  587. limit: 0
  588. };
  589. // Clear out current Op
  590. this.s.currentOp = null;
  591. return this.s.options.addToOperationsList(this, REMOVE, document);
  592. }
  593. /**
  594. * backwards compatability for deleteOne
  595. */
  596. removeOne() {
  597. return this.deleteOne();
  598. }
  599. /**
  600. * backwards compatability for delete
  601. */
  602. remove() {
  603. return this.delete();
  604. }
  605. }
  606. /**
  607. * Parent class to OrderedBulkOperation and UnorderedBulkOperation
  608. * @class
  609. */
  610. class BulkOperationBase {
  611. /**
  612. * Create a new OrderedBulkOperation or UnorderedBulkOperation instance (INTERNAL TYPE, do not instantiate directly)
  613. * @class
  614. * @property {number} length Get the number of operations in the bulk.
  615. * @return {OrderedBulkOperation|UnordedBulkOperation}
  616. */
  617. constructor(topology, collection, options, isOrdered) {
  618. // determine whether bulkOperation is ordered or unordered
  619. this.isOrdered = isOrdered;
  620. options = options == null ? {} : options;
  621. // TODO Bring from driver information in isMaster
  622. // Get the namespace for the write operations
  623. const namespace = collection.collectionName;
  624. // Used to mark operation as executed
  625. const executed = false;
  626. // Current item
  627. const currentOp = null;
  628. // Handle to the bson serializer, used to calculate running sizes
  629. const bson = topology.bson;
  630. // Set max byte size
  631. const isMaster = topology.lastIsMaster();
  632. const maxBatchSizeBytes =
  633. isMaster && isMaster.maxBsonObjectSize ? isMaster.maxBsonObjectSize : 1024 * 1024 * 16;
  634. const maxWriteBatchSize =
  635. isMaster && isMaster.maxWriteBatchSize ? isMaster.maxWriteBatchSize : 1000;
  636. // Calculates the largest possible size of an Array key, represented as a BSON string
  637. // element. This calculation:
  638. // 1 byte for BSON type
  639. // # of bytes = length of (string representation of (maxWriteBatchSize - 1))
  640. // + 1 bytes for null terminator
  641. const maxKeySize = (maxWriteBatchSize - 1).toString(10).length + 2;
  642. // Final options for retryable writes and write concern
  643. let finalOptions = Object.assign({}, options);
  644. finalOptions = applyRetryableWrites(finalOptions, collection.s.db);
  645. finalOptions = applyWriteConcern(finalOptions, { collection: collection }, options);
  646. const writeConcern = finalOptions.writeConcern;
  647. // Get the promiseLibrary
  648. const promiseLibrary = options.promiseLibrary || Promise;
  649. // Final results
  650. const bulkResult = {
  651. ok: 1,
  652. writeErrors: [],
  653. writeConcernErrors: [],
  654. insertedIds: [],
  655. nInserted: 0,
  656. nUpserted: 0,
  657. nMatched: 0,
  658. nModified: 0,
  659. nRemoved: 0,
  660. upserted: []
  661. };
  662. // Internal state
  663. this.s = {
  664. // Final result
  665. bulkResult: bulkResult,
  666. // Current batch state
  667. currentBatch: null,
  668. currentIndex: 0,
  669. // ordered specific
  670. currentBatchSize: 0,
  671. currentBatchSizeBytes: 0,
  672. // unordered specific
  673. currentInsertBatch: null,
  674. currentUpdateBatch: null,
  675. currentRemoveBatch: null,
  676. batches: [],
  677. // Write concern
  678. writeConcern: writeConcern,
  679. // Max batch size options
  680. maxBatchSizeBytes: maxBatchSizeBytes,
  681. maxWriteBatchSize: maxWriteBatchSize,
  682. maxKeySize,
  683. // Namespace
  684. namespace: namespace,
  685. // BSON
  686. bson: bson,
  687. // Topology
  688. topology: topology,
  689. // Options
  690. options: finalOptions,
  691. // Current operation
  692. currentOp: currentOp,
  693. // Executed
  694. executed: executed,
  695. // Collection
  696. collection: collection,
  697. // Promise Library
  698. promiseLibrary: promiseLibrary,
  699. // Fundamental error
  700. err: null,
  701. // check keys
  702. checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : true
  703. };
  704. // bypass Validation
  705. if (options.bypassDocumentValidation === true) {
  706. this.s.bypassDocumentValidation = true;
  707. }
  708. }
  709. /**
  710. * Add a single insert document to the bulk operation
  711. *
  712. * @param {object} document the document to insert
  713. * @throws {MongoError}
  714. * @return {OrderedBulkOperation|UnorderedBulkOperation}
  715. */
  716. insert(document) {
  717. if (this.s.collection.s.db.options.forceServerObjectId !== true && document._id == null)
  718. document._id = new ObjectID();
  719. return this.s.options.addToOperationsList(this, INSERT, document);
  720. }
  721. /**
  722. * Initiate a find operation for an update/updateOne/remove/removeOne/replaceOne
  723. *
  724. * @method
  725. * @param {object} selector The selector for the bulk operation.
  726. * @throws {MongoError}
  727. */
  728. find(selector) {
  729. if (!selector) {
  730. throw toError('Bulk find operation must specify a selector');
  731. }
  732. // Save a current selector
  733. this.s.currentOp = {
  734. selector: selector
  735. };
  736. return new FindOperators(this);
  737. }
  738. /**
  739. * Raw performs the bulk operation
  740. *
  741. * @method
  742. * @param {object} op operation
  743. * @return {OrderedBulkOperation|UnorderedBulkOperation}
  744. */
  745. raw(op) {
  746. const key = Object.keys(op)[0];
  747. // Set up the force server object id
  748. const forceServerObjectId =
  749. typeof this.s.options.forceServerObjectId === 'boolean'
  750. ? this.s.options.forceServerObjectId
  751. : this.s.collection.s.db.options.forceServerObjectId;
  752. // Update operations
  753. if (
  754. (op.updateOne && op.updateOne.q) ||
  755. (op.updateMany && op.updateMany.q) ||
  756. (op.replaceOne && op.replaceOne.q)
  757. ) {
  758. op[key].multi = op.updateOne || op.replaceOne ? false : true;
  759. return this.s.options.addToOperationsList(this, UPDATE, op[key]);
  760. }
  761. // Crud spec update format
  762. if (op.updateOne || op.updateMany || op.replaceOne) {
  763. const multi = op.updateOne || op.replaceOne ? false : true;
  764. const operation = {
  765. q: op[key].filter,
  766. u: op[key].update || op[key].replacement,
  767. multi: multi
  768. };
  769. if (this.isOrdered) {
  770. operation.upsert = op[key].upsert ? true : false;
  771. if (op.collation) operation.collation = op.collation;
  772. } else {
  773. if (op[key].upsert) operation.upsert = true;
  774. }
  775. if (op[key].arrayFilters) operation.arrayFilters = op[key].arrayFilters;
  776. return this.s.options.addToOperationsList(this, UPDATE, operation);
  777. }
  778. // Remove operations
  779. if (
  780. op.removeOne ||
  781. op.removeMany ||
  782. (op.deleteOne && op.deleteOne.q) ||
  783. (op.deleteMany && op.deleteMany.q)
  784. ) {
  785. op[key].limit = op.removeOne ? 1 : 0;
  786. return this.s.options.addToOperationsList(this, REMOVE, op[key]);
  787. }
  788. // Crud spec delete operations, less efficient
  789. if (op.deleteOne || op.deleteMany) {
  790. const limit = op.deleteOne ? 1 : 0;
  791. const operation = { q: op[key].filter, limit: limit };
  792. if (this.isOrdered) {
  793. if (op.collation) operation.collation = op.collation;
  794. }
  795. return this.s.options.addToOperationsList(this, REMOVE, operation);
  796. }
  797. // Insert operations
  798. if (op.insertOne && op.insertOne.document == null) {
  799. if (forceServerObjectId !== true && op.insertOne._id == null)
  800. op.insertOne._id = new ObjectID();
  801. return this.s.options.addToOperationsList(this, INSERT, op.insertOne);
  802. } else if (op.insertOne && op.insertOne.document) {
  803. if (forceServerObjectId !== true && op.insertOne.document._id == null)
  804. op.insertOne.document._id = new ObjectID();
  805. return this.s.options.addToOperationsList(this, INSERT, op.insertOne.document);
  806. }
  807. if (op.insertMany) {
  808. for (let i = 0; i < op.insertMany.length; i++) {
  809. if (forceServerObjectId !== true && op.insertMany[i]._id == null)
  810. op.insertMany[i]._id = new ObjectID();
  811. this.s.options.addToOperationsList(this, INSERT, op.insertMany[i]);
  812. }
  813. return;
  814. }
  815. // No valid type of operation
  816. throw toError(
  817. 'bulkWrite only supports insertOne, insertMany, updateOne, updateMany, removeOne, removeMany, deleteOne, deleteMany'
  818. );
  819. }
  820. _handleEarlyError(err, callback) {
  821. if (typeof callback === 'function') {
  822. callback(err, null);
  823. return;
  824. }
  825. return this.s.promiseLibrary.reject(err);
  826. }
  827. /**
  828. * Execute next write command in a chain
  829. *
  830. * @method
  831. * @param {class} bulk either OrderedBulkOperation or UnorderdBulkOperation
  832. * @param {object} writeConcern
  833. * @param {object} options
  834. * @param {function} callback
  835. */
  836. bulkExecute(_writeConcern, options, callback) {
  837. if (typeof options === 'function') (callback = options), (options = {});
  838. options = options || {};
  839. if (typeof _writeConcern === 'function') {
  840. callback = _writeConcern;
  841. } else if (_writeConcern && typeof _writeConcern === 'object') {
  842. this.s.writeConcern = _writeConcern;
  843. }
  844. if (this.s.executed) {
  845. const executedError = toError('batch cannot be re-executed');
  846. return this._handleEarlyError(executedError, callback);
  847. }
  848. // If we have current batch
  849. if (this.isOrdered) {
  850. if (this.s.currentBatch) this.s.batches.push(this.s.currentBatch);
  851. } else {
  852. if (this.s.currentInsertBatch) this.s.batches.push(this.s.currentInsertBatch);
  853. if (this.s.currentUpdateBatch) this.s.batches.push(this.s.currentUpdateBatch);
  854. if (this.s.currentRemoveBatch) this.s.batches.push(this.s.currentRemoveBatch);
  855. }
  856. // If we have no operations in the bulk raise an error
  857. if (this.s.batches.length === 0) {
  858. const emptyBatchError = toError('Invalid Operation, no operations specified');
  859. return this._handleEarlyError(emptyBatchError, callback);
  860. }
  861. return { options, callback };
  862. }
  863. /**
  864. * The callback format for results
  865. * @callback BulkOperationBase~resultCallback
  866. * @param {MongoError} error An error instance representing the error during the execution.
  867. * @param {BulkWriteResult} result The bulk write result.
  868. */
  869. /**
  870. * Execute the ordered bulk operation
  871. *
  872. * @method
  873. * @param {object} [options] Optional settings.
  874. * @param {(number|string)} [options.w] The write concern.
  875. * @param {number} [options.wtimeout] The write concern timeout.
  876. * @param {boolean} [options.j=false] Specify a journal write concern.
  877. * @param {boolean} [options.fsync=false] Specify a file sync write concern.
  878. * @param {BulkOperationBase~resultCallback} [callback] The result callback
  879. * @throws {MongoError} Throws error if the bulk object has already been executed
  880. * @throws {MongoError} Throws error if the bulk object does not have any operations
  881. * @return {Promise} returns Promise if no callback passed
  882. */
  883. execute(_writeConcern, options, callback) {
  884. const ret = this.bulkExecute(_writeConcern, options, callback);
  885. if (!ret || isPromiseLike(ret)) {
  886. return ret;
  887. }
  888. options = ret.options;
  889. callback = ret.callback;
  890. return executeOperation(this.s.topology, executeCommands, [this, options, callback]);
  891. }
  892. /**
  893. * Handles final options before executing command
  894. *
  895. * @param {object} config
  896. * @param {object} config.options
  897. * @param {number} config.batch
  898. * @param {function} config.resultHandler
  899. * @param {function} callback
  900. */
  901. finalOptionsHandler(config, callback) {
  902. const finalOptions = Object.assign({ ordered: this.isOrdered }, config.options);
  903. if (this.s.writeConcern != null) {
  904. finalOptions.writeConcern = this.s.writeConcern;
  905. }
  906. if (finalOptions.bypassDocumentValidation !== true) {
  907. delete finalOptions.bypassDocumentValidation;
  908. }
  909. // Set an operationIf if provided
  910. if (this.operationId) {
  911. config.resultHandler.operationId = this.operationId;
  912. }
  913. // Serialize functions
  914. if (this.s.options.serializeFunctions) {
  915. finalOptions.serializeFunctions = true;
  916. }
  917. // Ignore undefined
  918. if (this.s.options.ignoreUndefined) {
  919. finalOptions.ignoreUndefined = true;
  920. }
  921. // Is the bypassDocumentValidation options specific
  922. if (this.s.bypassDocumentValidation === true) {
  923. finalOptions.bypassDocumentValidation = true;
  924. }
  925. // Is the checkKeys option disabled
  926. if (this.s.checkKeys === false) {
  927. finalOptions.checkKeys = false;
  928. }
  929. if (finalOptions.retryWrites) {
  930. if (config.batch.batchType === UPDATE) {
  931. finalOptions.retryWrites =
  932. finalOptions.retryWrites && !config.batch.operations.some(op => op.multi);
  933. }
  934. if (config.batch.batchType === REMOVE) {
  935. finalOptions.retryWrites =
  936. finalOptions.retryWrites && !config.batch.operations.some(op => op.limit === 0);
  937. }
  938. }
  939. try {
  940. if (config.batch.batchType === INSERT) {
  941. this.s.topology.insert(
  942. this.s.collection.namespace,
  943. config.batch.operations,
  944. finalOptions,
  945. config.resultHandler
  946. );
  947. } else if (config.batch.batchType === UPDATE) {
  948. this.s.topology.update(
  949. this.s.collection.namespace,
  950. config.batch.operations,
  951. finalOptions,
  952. config.resultHandler
  953. );
  954. } else if (config.batch.batchType === REMOVE) {
  955. this.s.topology.remove(
  956. this.s.collection.namespace,
  957. config.batch.operations,
  958. finalOptions,
  959. config.resultHandler
  960. );
  961. }
  962. } catch (err) {
  963. // Force top level error
  964. err.ok = 0;
  965. // Merge top level error and return
  966. handleCallback(callback, null, mergeBatchResults(config.batch, this.s.bulkResult, err, null));
  967. }
  968. }
  969. /**
  970. * Handles the write error before executing commands
  971. *
  972. * @param {function} callback
  973. * @param {BulkWriteResult} writeResult
  974. * @param {class} self either OrderedBulkOperation or UnorderdBulkOperation
  975. */
  976. handleWriteError(callback, writeResult) {
  977. if (this.s.bulkResult.writeErrors.length > 0) {
  978. if (this.s.bulkResult.writeErrors.length === 1) {
  979. handleCallback(
  980. callback,
  981. new BulkWriteError(toError(this.s.bulkResult.writeErrors[0]), writeResult),
  982. null
  983. );
  984. return true;
  985. }
  986. handleCallback(
  987. callback,
  988. new BulkWriteError(
  989. toError({
  990. message: 'write operation failed',
  991. code: this.s.bulkResult.writeErrors[0].code,
  992. writeErrors: this.s.bulkResult.writeErrors
  993. }),
  994. writeResult
  995. ),
  996. null
  997. );
  998. return true;
  999. } else if (writeResult.getWriteConcernError()) {
  1000. handleCallback(
  1001. callback,
  1002. new BulkWriteError(toError(writeResult.getWriteConcernError()), writeResult),
  1003. null
  1004. );
  1005. return true;
  1006. }
  1007. }
  1008. }
  1009. Object.defineProperty(BulkOperationBase.prototype, 'length', {
  1010. enumerable: true,
  1011. get: function() {
  1012. return this.s.currentIndex;
  1013. }
  1014. });
  1015. // Exports symbols
  1016. module.exports = {
  1017. Batch,
  1018. BulkOperationBase,
  1019. bson,
  1020. INSERT: INSERT,
  1021. UPDATE: UPDATE,
  1022. REMOVE: REMOVE
  1023. };