Merge remote-tracking branch 'origin/master' into NODE-2757/add-collation-to-bulk-find-operators

Author: emadum

package.json

@@ -27,8 +27,7 @@
   },
   "dependencies": {
     "bson": "^4.0.4",
-    "denque": "^1.4.1",
-    "lodash": "^4.17.20"
+    "denque": "^1.4.1"
   },
   "devDependencies": {
     "@istanbuljs/nyc-config-typescript": "^1.0.1",
@@ -38,7 +37,6 @@
     "@types/bl": "^2.1.0",
     "@types/bson": "^4.0.2",
     "@types/kerberos": "^1.1.0",
-    "@types/lodash": "^4.14.164",
     "@types/node": "^14.6.4",
     "@types/saslprep": "^1.0.0",
     "@typescript-eslint/eslint-plugin": "^3.10.0",

src/bulk/common.ts

@@ -13,8 +13,8 @@ import {
 } from '../utils';
 import { executeOperation } from '../operations/execute_operation';
 import { InsertOperation } from '../operations/insert';
-import { UpdateOperation } from '../operations/update';
-import { DeleteOperation } from '../operations/delete';
+import { UpdateOperation, UpdateStatement } from '../operations/update';
+import { DeleteOperation, DeleteStatement } from '../operations/delete';
 import { WriteConcern } from '../write_concern';
 import type { Collection } from '../collection';
 import type { Topology } from '../sdam/topology';
@@ -28,7 +28,7 @@ const WRITE_CONCERN_ERROR = 64;
 export const BatchType = {
   INSERT: 1,
   UPDATE: 2,
-  REMOVE: 3
+  DELETE: 3
 } as const;
 
 /** @public */
@@ -139,12 +139,12 @@ export interface BulkResult {
  *
  * @public
  */
-export class Batch {
+export class Batch<T = Document> {
   originalZeroIndex: number;
   currentIndex: number;
   originalIndexes: number[];
   batchType: BatchTypeId;
-  operations: Document[];
+  operations: T[];
   size: number;
   sizeBytes: number;
 
@@ -483,12 +483,12 @@ function mergeBatchResults(
   }
 
   // If we have an insert Batch type
-  if (batch.batchType === BatchType.INSERT && result.n) {
+  if (isInsertBatch(batch) && result.n) {
     bulkResult.nInserted = bulkResult.nInserted + result.n;
   }
 
   // If we have an insert Batch type
-  if (batch.batchType === BatchType.REMOVE && result.n) {
+  if (isDeleteBatch(batch) && result.n) {
     bulkResult.nRemoved = bulkResult.nRemoved + result.n;
   }
 
@@ -514,7 +514,7 @@ function mergeBatchResults(
   }
 
   // If we have an update Batch type
-  if (batch.batchType === BatchType.UPDATE && result.n) {
+  if (isUpdateBatch(batch) && result.n) {
     const nModified = result.nModified;
     bulkResult.nUpserted = bulkResult.nUpserted + nUpserted;
     bulkResult.nMatched = bulkResult.nMatched + (result.n - nUpserted);
@@ -605,30 +605,30 @@ function executeCommands(
   }
 
   if (finalOptions.retryWrites) {
-    if (batch.batchType === BatchType.UPDATE) {
+    if (isUpdateBatch(batch)) {
       finalOptions.retryWrites = finalOptions.retryWrites && !batch.operations.some(op => op.multi);
     }
 
-    if (batch.batchType === BatchType.REMOVE) {
+    if (isDeleteBatch(batch)) {
       finalOptions.retryWrites =
         finalOptions.retryWrites && !batch.operations.some(op => op.limit === 0);
     }
   }
 
   try {
-    if (batch.batchType === BatchType.INSERT) {
+    if (isInsertBatch(batch)) {
       executeOperation(
         bulkOperation.s.topology,
         new InsertOperation(bulkOperation.s.namespace, batch.operations, finalOptions),
         resultHandler
       );
-    } else if (batch.batchType === BatchType.UPDATE) {
+    } else if (isUpdateBatch(batch)) {
       executeOperation(
         bulkOperation.s.topology,
         new UpdateOperation(bulkOperation.s.namespace, batch.operations, finalOptions),
         resultHandler
       );
-    } else if (batch.batchType === BatchType.REMOVE) {
+    } else if (isDeleteBatch(batch)) {
       executeOperation(
         bulkOperation.s.topology,
         new DeleteOperation(bulkOperation.s.namespace, batch.operations, finalOptions),
@@ -811,12 +811,12 @@ export class FindOperators {
 
   /** Add a delete one operation to the bulk operation */
   deleteOne(): BulkOperationBase {
-    return this.bulkOperation.addToOperationsList(BatchType.REMOVE, this.makeDeleteDocument(1));
+    return this.bulkOperation.addToOperationsList(BatchType.DELETE, this.makeDeleteDocument(1));
   }
 
   /** Add a delete many operation to the bulk operation */
   delete(): BulkOperationBase {
-    return this.bulkOperation.addToOperationsList(BatchType.REMOVE, this.makeDeleteDocument(0));
+    return this.bulkOperation.addToOperationsList(BatchType.DELETE, this.makeDeleteDocument(0));
   }
 
   removeOne(): BulkOperationBase {
@@ -1059,7 +1059,7 @@ export abstract class BulkOperationBase {
    * bulkOp.find({ h: 8 }).delete();
    *
    * // Add a replaceOne
-   * bulkOp.find({ i: 9 }).replaceOne({ j: 10 });
+   * bulkOp.find({ i: 9 }).replaceOne({writeConcern: { j: 10 }});
    *
    * // Update using a pipeline (requires Mongodb 4.2 or higher)
    * bulk.find({ k: 11, y: { $exists: true }, z: { $exists: true } }).updateOne([
@@ -1143,28 +1143,28 @@ export abstract class BulkOperationBase {
 
     if ('removeOne' in op) {
       return this.addToOperationsList(
-        BatchType.REMOVE,
+        BatchType.DELETE,
         makeDeleteStatement(this.s.topology, op.removeOne, false)
       );
     }
 
     if ('removeMany' in op) {
       return this.addToOperationsList(
-        BatchType.REMOVE,
+        BatchType.DELETE,
         makeDeleteStatement(this.s.topology, op.removeMany, true)
       );
     }
 
     if ('deleteOne' in op) {
       return this.addToOperationsList(
-        BatchType.REMOVE,
+        BatchType.DELETE,
         makeDeleteStatement(this.s.topology, op.deleteOne, false)
       );
     }
 
     if ('deleteMany' in op) {
       return this.addToOperationsList(
-        BatchType.REMOVE,
+        BatchType.DELETE,
         makeDeleteStatement(this.s.topology, op.deleteMany, true)
       );
     }
@@ -1303,24 +1303,6 @@ function shouldForceServerObjectId(bulkOperation: BulkOperationBase): boolean {
   return false;
 }
 
-/** @public */
-export interface UpdateStatement {
-  /** The query that matches documents to update. */
-  q: Document;
-  /** The modifications to apply. */
-  u: Document | Document[];
-  /**  If true, perform an insert if no documents match the query. */
-  upsert?: boolean;
-  /** If true, updates all documents that meet the query criteria. */
-  multi?: boolean;
-  /** Specifies the collation to use for the operation. */
-  collation?: CollationOptions;
-  /** An array of filter documents that determines which array elements to modify for an update operation on an array field. */
-  arrayFilters?: Document[];
-  /** A document or string that specifies the index to use to support the query predicate. */
-  hint?: Hint;
-}
-
 function makeUpdateStatement(
   topology: Topology,
   model: ReplaceOneModel | UpdateOneModel | UpdateManyModel,
@@ -1370,18 +1352,6 @@ function isUpdateStatement(model: Document): model is UpdateStatement {
   return 'q' in model;
 }
 
-/** @public */
-export interface DeleteStatement {
-  /** The query that matches documents to delete. */
-  q: Document;
-  /** The number of matching documents to delete. */
-  limit: number;
-  /** Specifies the collation to use for the operation. */
-  collation?: CollationOptions;
-  /** A document or string that specifies the index to use to support the query predicate. */
-  hint?: Hint;
-}
-
 function makeDeleteStatement(
   topology: Topology,
   model: DeleteOneModel | DeleteManyModel,
@@ -1420,3 +1390,15 @@ function makeDeleteStatement(
 function isDeleteStatement(model: Document): model is DeleteStatement {
   return 'q' in model;
 }
+
+function isInsertBatch(batch: Batch): boolean {
+  return batch.batchType === BatchType.INSERT;
+}
+
+function isUpdateBatch(batch: Batch): batch is Batch<UpdateStatement> {
+  return batch.batchType === BatchType.UPDATE;
+}
+
+function isDeleteBatch(batch: Batch): batch is Batch<DeleteStatement> {
+  return batch.batchType === BatchType.DELETE;
+}

src/bulk/ordered.ts

@@ -1,15 +1,9 @@
 import * as BSON from '../bson';
-import {
-  BulkOperationBase,
-  Batch,
-  BatchType,
-  BulkWriteOptions,
-  UpdateStatement,
-  DeleteStatement,
-  BatchTypeId
-} from './common';
+import { BulkOperationBase, Batch, BatchType, BulkWriteOptions, BatchTypeId } from './common';
 import type { Document } from '../bson';
 import type { Collection } from '../collection';
+import type { UpdateStatement } from '../operations/update';
+import type { DeleteStatement } from '../operations/delete';
 
 /** @public */
 export class OrderedBulkOperation extends BulkOperationBase {

src/bulk/unordered.ts

@@ -4,14 +4,14 @@ import {
   Batch,
   BatchType,
   BulkWriteOptions,
-  UpdateStatement,
-  DeleteStatement,
   BulkWriteResult,
   BatchTypeId
 } from './common';
 import type { Callback } from '../utils';
 import type { Document } from '../bson';
 import type { Collection } from '../collection';
+import type { UpdateStatement } from '../operations/update';
+import type { DeleteStatement } from '../operations/delete';
 
 /** @public */
 export class UnorderedBulkOperation extends BulkOperationBase {
@@ -52,7 +52,7 @@ export class UnorderedBulkOperation extends BulkOperationBase {
       this.s.currentBatch = this.s.currentInsertBatch;
     } else if (batchType === BatchType.UPDATE) {
       this.s.currentBatch = this.s.currentUpdateBatch;
-    } else if (batchType === BatchType.REMOVE) {
+    } else if (batchType === BatchType.DELETE) {
       this.s.currentBatch = this.s.currentRemoveBatch;
     }
 
@@ -99,7 +99,7 @@ export class UnorderedBulkOperation extends BulkOperationBase {
       });
     } else if (batchType === BatchType.UPDATE) {
       this.s.currentUpdateBatch = this.s.currentBatch;
-    } else if (batchType === BatchType.REMOVE) {
+    } else if (batchType === BatchType.DELETE) {
       this.s.currentRemoveBatch = this.s.currentBatch;
     }
 

src/change_stream.ts

@@ -46,6 +46,7 @@ const CHANGE_DOMAIN_TYPES = {
 const NO_RESUME_TOKEN_ERROR = new MongoError(
   'A change stream document has been received that lacks a resume token (_id).'
 );
+const NO_CURSOR_ERROR = new MongoError('ChangeStream has no cursor');
 const CHANGESTREAM_CLOSED_ERROR = new MongoError('ChangeStream is closed');
 
 /** @public */
@@ -287,8 +288,7 @@ export class ChangeStream extends EventEmitter {
   next(callback?: Callback): Promise<void> | void {
     return maybePromise(callback, cb => {
       getCursor(this, (err, cursor) => {
-        if (err) return cb(err); // failed to resume, raise an error
-        if (!cursor) return cb(new MongoError('Cursor is undefined'));
+        if (err || !cursor) return cb(err); // failed to resume, raise an error
         cursor.next((error, change) => {
           if (error) {
             this[kResumeQueue].push(() => this.next(cb));
@@ -330,11 +330,23 @@ export class ChangeStream extends EventEmitter {
    */
   stream(options?: CursorStreamOptions): Readable {
     this.streamOptions = options;
-    if (!this.cursor) {
-      throw new MongoError('ChangeStream has no cursor, unable to stream');
-    }
+    if (!this.cursor) throw NO_CURSOR_ERROR;
     return this.cursor.stream(options);
   }
+
+  /**
+   * Try to get the next available document from the Change Stream's cursor or `null` if an empty batch is returned
+   */
+  tryNext(): Promise<Document | null>;
+  tryNext(callback: Callback<Document | null>): void;
+  tryNext(callback?: Callback<Document | null>): Promise<Document | null> | void {
+    return maybePromise(callback, cb => {
+      getCursor(this, (err, cursor) => {
+        if (err || !cursor) return cb(err); // failed to resume, raise an error
+        return cursor.tryNext(cb);
+      });
+    });
+  }
 }
 
 /** @internal */
@@ -707,11 +719,16 @@ function getCursor(changeStream: ChangeStream, callback: Callback<ChangeStreamCu
 function processResumeQueue(changeStream: ChangeStream, err?: Error) {
   while (changeStream[kResumeQueue].length) {
     const request = changeStream[kResumeQueue].pop();
-    if (changeStream[kClosed] && !err) {
-      request(CHANGESTREAM_CLOSED_ERROR);
-      return;
+    if (!err) {
+      if (changeStream[kClosed]) {
+        request(CHANGESTREAM_CLOSED_ERROR);
+        return;
+      }
+      if (!changeStream.cursor) {
+        request(NO_CURSOR_ERROR);
+        return;
+      }
     }
-
     request(err, changeStream.cursor);
   }
 }