refactor(NODE-4631): change_stream, gridfs to use maybeCallback

Author: nbbeeken

src/bulk/common.ts

@@ -1346,13 +1346,13 @@ function handleEarlyError(
   err?: AnyError,
   callback?: Callback<BulkWriteResult>
 ): Promise<BulkWriteResult> | void {
-  const Promise = PromiseProvider.get();
   if (typeof callback === 'function') {
     callback(err);
     return;
   }
 
-  return Promise.reject(err);
+  const PromiseConstructor = PromiseProvider.get() ?? Promise;
+  return PromiseConstructor.reject(err);
 }
 
 function shouldForceServerObjectId(bulkOperation: BulkOperationBase): boolean {

src/change_stream.ts

@@ -20,7 +20,7 @@ import type { AggregateOptions } from './operations/aggregate';
 import type { CollationOptions, OperationParent } from './operations/command';
 import type { ReadPreference } from './read_preference';
 import type { ServerSessionId } from './sessions';
-import { Callback, filterOptions, getTopology, maybePromise, MongoDBNamespace } from './utils';
+import { Callback, filterOptions, getTopology, maybeCallback, MongoDBNamespace } from './utils';
 
 /** @internal */
 const kCursorStream = Symbol('cursorStream');
@@ -649,29 +649,25 @@ export class ChangeStream<
   hasNext(callback: Callback<boolean>): void;
   hasNext(callback?: Callback): Promise<boolean> | void {
     this._setIsIterator();
-    // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
-    // Shadowing is intentional here.  We want to override the `callback` variable
-    // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
-    return maybePromise(callback, callback => {
-      (async () => {
+    return maybeCallback(async () => {
+      try {
+        const hasNext = await this.cursor.hasNext();
+        return hasNext;
+      } catch (error) {
         try {
+          await this._processErrorIteratorMode(error);
           const hasNext = await this.cursor.hasNext();
           return hasNext;
         } catch (error) {
           try {
-            await this._processErrorIteratorMode(error);
-            const hasNext = await this.cursor.hasNext();
-            return hasNext;
-          } catch (error) {
-            await this.close().catch(err => err);
-            throw error;
+            await this.close();
+          } catch {
+            // We are not concerned with errors from close()
           }
+          throw error;
         }
-      })().then(
-        hasNext => callback(undefined, hasNext),
-        error => callback(error)
-      );
-    });
+      }
+    }, callback);
   }
 
   /** Get the next available document from the Change Stream. */
@@ -680,31 +676,27 @@ export class ChangeStream<
   next(callback: Callback<TChange>): void;
   next(callback?: Callback<TChange>): Promise<TChange> | void {
     this._setIsIterator();
-    // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
-    // Shadowing is intentional here.  We want to override the `callback` variable
-    // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
-    return maybePromise(callback, callback => {
-      (async () => {
+    return maybeCallback(async () => {
+      try {
+        const change = await this.cursor.next();
+        const processedChange = this._processChange(change ?? null);
+        return processedChange;
+      } catch (error) {
         try {
+          await this._processErrorIteratorMode(error);
           const change = await this.cursor.next();
           const processedChange = this._processChange(change ?? null);
           return processedChange;
         } catch (error) {
           try {
-            await this._processErrorIteratorMode(error);
-            const change = await this.cursor.next();
-            const processedChange = this._processChange(change ?? null);
-            return processedChange;
-          } catch (error) {
-            await this.close().catch(err => err);
-            throw error;
+            await this.close();
+          } catch {
+            // We are not concerned with errors from close()
           }
+          throw error;
         }
-      })().then(
-        change => callback(undefined, change),
-        error => callback(error)
-      );
-    });
+      }
+    }, callback);
   }
 
   /**
@@ -715,29 +707,21 @@ export class ChangeStream<
   tryNext(callback: Callback<Document | null>): void;
   tryNext(callback?: Callback<Document | null>): Promise<Document | null> | void {
     this._setIsIterator();
-    // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
-    // Shadowing is intentional here.  We want to override the `callback` variable
-    // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
-    return maybePromise(callback, callback => {
-      (async () => {
+    return maybeCallback(async () => {
+      try {
+        const change = await this.cursor.tryNext();
+        return change ?? null;
+      } catch (error) {
         try {
+          await this._processErrorIteratorMode(error);
           const change = await this.cursor.tryNext();
           return change ?? null;
         } catch (error) {
-          try {
-            await this._processErrorIteratorMode(error);
-            const change = await this.cursor.tryNext();
-            return change ?? null;
-          } catch (error) {
-            await this.close().catch(err => err);
-            throw error;
-          }
+          await this.close().catch(err => err);
+          throw error;
         }
-      })().then(
-        change => callback(undefined, change),
-        error => callback(error)
-      );
-    });
+      }
+    }, callback);
   }
 
   /** Is the cursor closed */
@@ -752,13 +736,14 @@ export class ChangeStream<
   close(callback?: Callback): Promise<void> | void {
     this[kClosed] = true;
 
-    return maybePromise(callback, cb => {
+    return maybeCallback(async () => {
       const cursor = this.cursor;
-      return cursor.close(err => {
+      try {
+        await cursor.close();
+      } finally {
         this._endStream();
-        return cb(err);
-      });
-    });
+      }
+    }, callback);
   }
 
   /**

src/gridfs/index.ts

@@ -7,7 +7,7 @@ import type { Logger } from '../logger';
 import { Filter, TypedEventEmitter } from '../mongo_types';
 import type { ReadPreference } from '../read_preference';
 import type { Sort } from '../sort';
-import { Callback, maybePromise } from '../utils';
+import { Callback, maybeCallback } from '../utils';
 import { WriteConcern, WriteConcernOptions } from '../write_concern';
 import type { FindOptions } from './../operations/find';
 import {
@@ -144,28 +144,17 @@ export class GridFSBucket extends TypedEventEmitter<GridFSBucketEvents> {
   /** @deprecated Callbacks are deprecated and will be removed in the next major version. See [mongodb-legacy](https://github.com/mongodb-js/nodejs-mongodb-legacy) for migration assistance */
   delete(id: ObjectId, callback: Callback<void>): void;
   delete(id: ObjectId, callback?: Callback<void>): Promise<void> | void {
-    return maybePromise(callback, callback => {
-      return this.s._filesCollection.deleteOne({ _id: id }, (error, res) => {
-        if (error) {
-          return callback(error);
-        }
-
-        return this.s._chunksCollection.deleteMany({ files_id: id }, error => {
-          if (error) {
-            return callback(error);
-          }
-
-          // Delete orphaned chunks before returning FileNotFound
-          if (!res?.deletedCount) {
-            // TODO(NODE-3483): Replace with more appropriate error
-            // Consider creating new error MongoGridFSFileNotFoundError
-            return callback(new MongoRuntimeError(`File not found for id ${id}`));
-          }
-
-          return callback();
-        });
-      });
-    });
+    return maybeCallback(async () => {
+      const { deletedCount } = await this.s._filesCollection.deleteOne({ _id: id });
+      await this.s._chunksCollection.deleteMany({ files_id: id });
+
+      // Delete orphaned chunks before returning FileNotFound
+      if (deletedCount === 0) {
+        // TODO(NODE-3483): Replace with more appropriate error
+        // Consider creating new error MongoGridFSFileNotFoundError
+        throw new MongoRuntimeError(`File not found for id ${id}`);
+      }
+    }, callback);
   }
 
   /** Convenience wrapper around find on the files collection */
@@ -215,42 +204,25 @@ export class GridFSBucket extends TypedEventEmitter<GridFSBucketEvents> {
   /** @deprecated Callbacks are deprecated and will be removed in the next major version. See [mongodb-legacy](https://github.com/mongodb-js/nodejs-mongodb-legacy) for migration assistance */
   rename(id: ObjectId, filename: string, callback: Callback<void>): void;
   rename(id: ObjectId, filename: string, callback?: Callback<void>): Promise<void> | void {
-    return maybePromise(callback, callback => {
+    return maybeCallback(async () => {
       const filter = { _id: id };
       const update = { $set: { filename } };
-      return this.s._filesCollection.updateOne(filter, update, (error?, res?) => {
-        if (error) {
-          return callback(error);
-        }
-
-        if (!res?.matchedCount) {
-          return callback(new MongoRuntimeError(`File with id ${id} not found`));
-        }
-
-        return callback();
-      });
-    });
+      const { matchedCount } = await this.s._filesCollection.updateOne(filter, update);
+      if (matchedCount === 0) {
+        throw new MongoRuntimeError(`File with id ${id} not found`);
+      }
+    }, callback);
   }
 
   /** Removes this bucket's files collection, followed by its chunks collection. */
   drop(): Promise<void>;
   /** @deprecated Callbacks are deprecated and will be removed in the next major version. See [mongodb-legacy](https://github.com/mongodb-js/nodejs-mongodb-legacy) for migration assistance */
   drop(callback: Callback<void>): void;
   drop(callback?: Callback<void>): Promise<void> | void {
-    return maybePromise(callback, callback => {
-      return this.s._filesCollection.drop(error => {
-        if (error) {
-          return callback(error);
-        }
-        return this.s._chunksCollection.drop(error => {
-          if (error) {
-            return callback(error);
-          }
-
-          return callback();
-        });
-      });
-    });
+    return maybeCallback(async () => {
+      await this.s._filesCollection.drop();
+      await this.s._chunksCollection.drop();
+    }, callback);
   }
 
   /** Get the Db scoped logger. */

src/gridfs/upload.ts

@@ -4,7 +4,7 @@ import type { Document } from '../bson';
 import { ObjectId } from '../bson';
 import type { Collection } from '../collection';
 import { AnyError, MongoAPIError, MONGODB_ERROR_CODES, MongoError } from '../error';
-import { Callback, maybePromise } from '../utils';
+import { Callback, maybeCallback } from '../utils';
 import type { WriteConcernOptions } from '../write_concern';
 import { WriteConcern } from './../write_concern';
 import type { GridFSFile } from './download';
@@ -149,20 +149,20 @@ export class GridFSBucketWriteStream extends Writable implements NodeJS.Writable
   /** @deprecated Callbacks are deprecated and will be removed in the next major version. See [mongodb-legacy](https://github.com/mongodb-js/nodejs-mongodb-legacy) for migration assistance */
   abort(callback: Callback<void>): void;
   abort(callback?: Callback<void>): Promise<void> | void {
-    return maybePromise(callback, callback => {
+    return maybeCallback(async () => {
       if (this.state.streamEnd) {
         // TODO(NODE-3485): Replace with MongoGridFSStreamClosed
-        return callback(new MongoAPIError('Cannot abort a stream that has already completed'));
+        throw new MongoAPIError('Cannot abort a stream that has already completed');
       }
 
       if (this.state.aborted) {
         // TODO(NODE-3485): Replace with MongoGridFSStreamClosed
-        return callback(new MongoAPIError('Cannot call abort() on a stream twice'));
+        throw new MongoAPIError('Cannot call abort() on a stream twice');
       }
 
       this.state.aborted = true;
-      this.chunks.deleteMany({ files_id: this.id }, error => callback(error));
-    });
+      await this.chunks.deleteMany({ files_id: this.id });
+    }, callback);
   }
 
   /**

src/mongo_client.ts

@@ -593,8 +593,12 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
         return mongoClient.connect();
       }
     } catch (error) {
-      if (callback) return callback(error);
-      else return PromiseProvider.get().reject(error);
+      if (callback) {
+        return callback(error);
+      }
+
+      const PromiseConstructor = PromiseProvider.get() ?? Promise;
+      return PromiseConstructor.reject(error);
     }
   }
 
@@ -645,9 +649,9 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
     }
 
     const session = this.startSession(options);
-    const Promise = PromiseProvider.get();
+    const PromiseConstructor = PromiseProvider.get() ?? Promise;
 
-    return Promise.resolve()
+    return PromiseConstructor.resolve()
       .then(() => withSessionCallback(session))
       .then(() => {
         // Do not return the result of callback

src/promise_provider.ts

@@ -4,11 +4,11 @@ import { MongoInvalidArgumentError } from './error';
 const kPromise = Symbol('promise');
 
 interface PromiseStore {
-  [kPromise]?: PromiseConstructor;
+  [kPromise]: PromiseConstructor | null;
 }
 
 const store: PromiseStore = {
-  [kPromise]: undefined
+  [kPromise]: null
 };
 
 /**
@@ -31,7 +31,13 @@ export class PromiseProvider {
    * Sets the promise library
    * @deprecated Setting a custom promise library is deprecated the next major version will use the global Promise constructor only.
    */
-  static set(lib: PromiseConstructor): void {
+  static set(lib: PromiseConstructor | null): void {
+    // eslint-disable-next-line no-restricted-syntax
+    if (lib === null) {
+      // Check explicitly against null since `.set()` (no args) should fall through to validate
+      store[kPromise] = null;
+      return;
+    }
     if (!PromiseProvider.validate(lib)) {
       // validate
       return;
@@ -43,9 +49,7 @@ export class PromiseProvider {
    * Get the stored promise library, or resolves passed in
    * @deprecated Setting a custom promise library is deprecated the next major version will use the global Promise constructor only.
    */
-  static get(): PromiseConstructor {
-    return store[kPromise] as PromiseConstructor;
+  static get(): PromiseConstructor | null {
+    return store[kPromise];
   }
 }
-
-PromiseProvider.set(global.Promise);

src/sessions.ts

@@ -609,14 +609,14 @@ function attemptTransaction<TSchema>(
   fn: WithTransactionCallback<TSchema>,
   options?: TransactionOptions
 ): Promise<any> {
-  const Promise = PromiseProvider.get();
   session.startTransaction(options);
 
   let promise;
   try {
     promise = fn(session);
   } catch (err) {
-    promise = Promise.reject(err);
+    const PromiseConstructor = PromiseProvider.get() ?? Promise;
+    promise = PromiseConstructor.reject(err);
   }
 
   if (!isPromiseLike(promise)) {