refactor(NODE-4632): async await in MongoClient, ClientSession, and AbstractCursor

Author: nbbeeken

src/change_stream.ts

@@ -239,7 +239,7 @@ export interface ChangeStreamInsertDocument<TSchema extends Document = Document>
   operationType: 'insert';
   /** This key will contain the document being inserted */
   fullDocument: TSchema;
-  /** Namespace the insert event occured on */
+  /** Namespace the insert event occurred on */
   ns: ChangeStreamNameSpace;
 }
 
@@ -262,7 +262,7 @@ export interface ChangeStreamUpdateDocument<TSchema extends Document = Document>
   fullDocument?: TSchema;
   /** Contains a description of updated and removed fields in this operation */
   updateDescription: UpdateDescription<TSchema>;
-  /** Namespace the update event occured on */
+  /** Namespace the update event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -285,7 +285,7 @@ export interface ChangeStreamReplaceDocument<TSchema extends Document = Document
   operationType: 'replace';
   /** The fullDocument of a replace event represents the document after the insert of the replacement document */
   fullDocument: TSchema;
-  /** Namespace the replace event occured on */
+  /** Namespace the replace event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -307,7 +307,7 @@ export interface ChangeStreamDeleteDocument<TSchema extends Document = Document>
     ChangeStreamDocumentCollectionUUID {
   /** Describes the type of operation represented in this change notification */
   operationType: 'delete';
-  /** Namespace the delete event occured on */
+  /** Namespace the delete event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -328,7 +328,7 @@ export interface ChangeStreamDropDocument
     ChangeStreamDocumentCollectionUUID {
   /** Describes the type of operation represented in this change notification */
   operationType: 'drop';
-  /** Namespace the drop event occured on */
+  /** Namespace the drop event occurred on */
   ns: ChangeStreamNameSpace;
 }
 
@@ -343,7 +343,7 @@ export interface ChangeStreamRenameDocument
   operationType: 'rename';
   /** The new name for the `ns.coll` collection */
   to: { db: string; coll: string };
-  /** The "from" namespace that the rename occured on */
+  /** The "from" namespace that the rename occurred on */
   ns: ChangeStreamNameSpace;
 }
 

src/collection.ts

@@ -1478,7 +1478,7 @@ export class Collection<TSchema extends Document = Document> {
    * Create a new Change Stream, watching for new changes (insertions, updates, replacements, deletions, and invalidations) in this collection.
    *
    * @remarks
-   * watch() accepts two generic arguments for distinct usecases:
+   * watch() accepts two generic arguments for distinct use cases:
    * - The first is to override the schema that may be defined for this specific collection
    * - The second is to override the shape of the change stream document entirely, if it is not provided the type will default to ChangeStreamDocument of the first argument
    * @example
@@ -1603,7 +1603,7 @@ export class Collection<TSchema extends Document = Document> {
    *
    * @throws MongoNotConnectedError
    * @remarks
-   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implemenation.
+   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implementation.
    * However, `collection.bulkWrite()` provides an equivalent API that does not require prior connecting.
    */
   initializeUnorderedBulkOp(options?: BulkWriteOptions): UnorderedBulkOperation {
@@ -1615,7 +1615,7 @@ export class Collection<TSchema extends Document = Document> {
    *
    * @throws MongoNotConnectedError
    * @remarks
-   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implemenation.
+   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implementation.
    * However, `collection.bulkWrite()` provides an equivalent API that does not require prior connecting.
    */
   initializeOrderedBulkOp(options?: BulkWriteOptions): OrderedBulkOperation {

src/connection_string.ts

@@ -2,6 +2,7 @@ import * as dns from 'dns';
 import * as fs from 'fs';
 import ConnectionString from 'mongodb-connection-string-url';
 import { URLSearchParams } from 'url';
+import { promisify } from 'util';
 
 import type { Document } from './bson';
 import { MongoCredentials } from './cmap/auth/mongo_credentials';
@@ -30,7 +31,6 @@ import { ReadPreference, ReadPreferenceMode } from './read_preference';
 import type { TagSet } from './sdam/server_description';
 import {
   AnyOptions,
-  Callback,
   DEFAULT_PK_FACTORY,
   emitWarning,
   emitWarningOnce,
@@ -70,97 +70,91 @@ function matchesParentDomain(srvAddress: string, parentDomain: string): boolean
  * @param uri - The connection string to parse
  * @param options - Optional user provided connection string options
  */
-export function resolveSRVRecord(options: MongoOptions, callback: Callback<HostAddress[]>): void {
+export async function resolveSRVRecord(options: MongoOptions): Promise<HostAddress[]> {
+  // We need to do this here to make sinon tests work :(
+  const dnsResolveSrv = promisify(dns.resolveSrv);
+  const dnsResolveTxt = promisify(dns.resolveTxt);
+
   if (typeof options.srvHost !== 'string') {
-    return callback(new MongoAPIError('Option "srvHost" must not be empty'));
+    throw new MongoAPIError('Option "srvHost" must not be empty');
   }
 
   if (options.srvHost.split('.').length < 3) {
     // TODO(NODE-3484): Replace with MongoConnectionStringError
-    return callback(new MongoAPIError('URI must include hostname, domain name, and tld'));
+    throw new MongoAPIError('URI must include hostname, domain name, and tld');
   }
 
   // Resolve the SRV record and use the result as the list of hosts to connect to.
   const lookupAddress = options.srvHost;
-  dns.resolveSrv(`_${options.srvServiceName}._tcp.${lookupAddress}`, (err, addresses) => {
-    if (err) return callback(err);
+  const addresses = await dnsResolveSrv(`_${options.srvServiceName}._tcp.${lookupAddress}`);
 
-    if (addresses.length === 0) {
-      return callback(new MongoAPIError('No addresses found at host'));
-    }
+  if (addresses.length === 0) {
+    throw new MongoAPIError('No addresses found at host');
+  }
 
-    for (const { name } of addresses) {
-      if (!matchesParentDomain(name, lookupAddress)) {
-        return callback(new MongoAPIError('Server record does not share hostname with parent URI'));
-      }
+  for (const { name } of addresses) {
+    if (!matchesParentDomain(name, lookupAddress)) {
+      throw new MongoAPIError('Server record does not share hostname with parent URI');
     }
+  }
 
-    const hostAddresses = addresses.map(r =>
-      HostAddress.fromString(`${r.name}:${r.port ?? 27017}`)
-    );
+  const hostAddresses = addresses.map(r => HostAddress.fromString(`${r.name}:${r.port ?? 27017}`));
 
-    const lbError = validateLoadBalancedOptions(hostAddresses, options, true);
-    if (lbError) {
-      return callback(lbError);
+  validateLoadBalancedOptions(hostAddresses, options, true);
+
+  // Resolve TXT record and add options from there if they exist.
+  let record;
+  try {
+    record = await dnsResolveTxt(lookupAddress);
+  } catch (error) {
+    if (error.code !== 'ENODATA' && error.code !== 'ENOTFOUND') {
+      throw error;
     }
+    return hostAddresses;
+  }
 
-    // Resolve TXT record and add options from there if they exist.
-    dns.resolveTxt(lookupAddress, (err, record) => {
-      if (err) {
-        if (err.code !== 'ENODATA' && err.code !== 'ENOTFOUND') {
-          return callback(err);
-        }
-      } else {
-        if (record.length > 1) {
-          return callback(new MongoParseError('Multiple text records not allowed'));
-        }
+  if (record.length > 1) {
+    throw new MongoParseError('Multiple text records not allowed');
+  }
 
-        const txtRecordOptions = new URLSearchParams(record[0].join(''));
-        const txtRecordOptionKeys = [...txtRecordOptions.keys()];
-        if (txtRecordOptionKeys.some(key => !VALID_TXT_RECORDS.includes(key))) {
-          return callback(
-            new MongoParseError(`Text record may only set any of: ${VALID_TXT_RECORDS.join(', ')}`)
-          );
-        }
+  const txtRecordOptions = new URLSearchParams(record[0].join(''));
+  const txtRecordOptionKeys = [...txtRecordOptions.keys()];
+  if (txtRecordOptionKeys.some(key => !VALID_TXT_RECORDS.includes(key))) {
+    throw new MongoParseError(`Text record may only set any of: ${VALID_TXT_RECORDS.join(', ')}`);
+  }
 
-        if (VALID_TXT_RECORDS.some(option => txtRecordOptions.get(option) === '')) {
-          return callback(new MongoParseError('Cannot have empty URI params in DNS TXT Record'));
-        }
+  if (VALID_TXT_RECORDS.some(option => txtRecordOptions.get(option) === '')) {
+    throw new MongoParseError('Cannot have empty URI params in DNS TXT Record');
+  }
 
-        const source = txtRecordOptions.get('authSource') ?? undefined;
-        const replicaSet = txtRecordOptions.get('replicaSet') ?? undefined;
-        const loadBalanced = txtRecordOptions.get('loadBalanced') ?? undefined;
-
-        if (
-          !options.userSpecifiedAuthSource &&
-          source &&
-          options.credentials &&
-          !AUTH_MECHS_AUTH_SRC_EXTERNAL.has(options.credentials.mechanism)
-        ) {
-          options.credentials = MongoCredentials.merge(options.credentials, { source });
-        }
+  const source = txtRecordOptions.get('authSource') ?? undefined;
+  const replicaSet = txtRecordOptions.get('replicaSet') ?? undefined;
+  const loadBalanced = txtRecordOptions.get('loadBalanced') ?? undefined;
 
-        if (!options.userSpecifiedReplicaSet && replicaSet) {
-          options.replicaSet = replicaSet;
-        }
+  if (
+    !options.userSpecifiedAuthSource &&
+    source &&
+    options.credentials &&
+    !AUTH_MECHS_AUTH_SRC_EXTERNAL.has(options.credentials.mechanism)
+  ) {
+    options.credentials = MongoCredentials.merge(options.credentials, { source });
+  }
 
-        if (loadBalanced === 'true') {
-          options.loadBalanced = true;
-        }
+  if (!options.userSpecifiedReplicaSet && replicaSet) {
+    options.replicaSet = replicaSet;
+  }
 
-        if (options.replicaSet && options.srvMaxHosts > 0) {
-          return callback(new MongoParseError('Cannot combine replicaSet option with srvMaxHosts'));
-        }
+  if (loadBalanced === 'true') {
+    options.loadBalanced = true;
+  }
 
-        const lbError = validateLoadBalancedOptions(hostAddresses, options, true);
-        if (lbError) {
-          return callback(lbError);
-        }
-      }
+  if (options.replicaSet && options.srvMaxHosts > 0) {
+    throw new MongoParseError('Cannot combine replicaSet option with srvMaxHosts');
+  }
+
+  validateLoadBalancedOptions(hostAddresses, options, true);
 
-      callback(undefined, hostAddresses);
-    });
-  });
+  return hostAddresses;
 }
 
 /**
@@ -442,10 +436,8 @@ export function parseOptions(
     PromiseProvider.set(options.promiseLibrary);
   }
 
-  const lbError = validateLoadBalancedOptions(hosts, mongoOptions, isSRV);
-  if (lbError) {
-    throw lbError;
-  }
+  validateLoadBalancedOptions(hosts, mongoOptions, isSRV);
+
   if (mongoClient && mongoOptions.autoEncryption) {
     Encrypter.checkForMongoCrypt();
     mongoOptions.encrypter = new Encrypter(mongoClient, uri, options);
@@ -526,20 +518,20 @@ function validateLoadBalancedOptions(
   hosts: HostAddress[] | string[],
   mongoOptions: MongoOptions,
   isSrv: boolean
-): MongoParseError | undefined {
+): void {
   if (mongoOptions.loadBalanced) {
     if (hosts.length > 1) {
-      return new MongoParseError(LB_SINGLE_HOST_ERROR);
+      throw new MongoParseError(LB_SINGLE_HOST_ERROR);
     }
     if (mongoOptions.replicaSet) {
-      return new MongoParseError(LB_REPLICA_SET_ERROR);
+      throw new MongoParseError(LB_REPLICA_SET_ERROR);
     }
     if (mongoOptions.directConnection) {
-      return new MongoParseError(LB_DIRECT_CONNECTION_ERROR);
+      throw new MongoParseError(LB_DIRECT_CONNECTION_ERROR);
     }
 
     if (isSrv && mongoOptions.srvMaxHosts > 0) {
-      return new MongoParseError('Cannot limit srv hosts with loadBalanced enabled');
+      throw new MongoParseError('Cannot limit srv hosts with loadBalanced enabled');
     }
   }
   return;