Merge branch 'master' into NODE-2458/cursor-sort

Author: Thomas Reggi

src/change_stream.ts

@@ -262,11 +262,6 @@ export class ChangeStream extends EventEmitter {
     });
   }
 
-  /** @internal */
-  triggerError(error: AnyError): void {
-    processError(this, error);
-  }
-
   /** @internal */
   get cursorStream(): CursorStream | undefined {
     return this[kCursorStream];
@@ -568,12 +563,19 @@ function processNewChange(
   change: ChangeStreamDocument,
   callback?: Callback
 ) {
-  if (changeStream.closed) return callback && callback(CHANGESTREAM_CLOSED_ERROR);
+  if (changeStream.closed) {
+    if (callback) callback(CHANGESTREAM_CLOSED_ERROR);
+    return;
+  }
 
   // a null change means the cursor has been notified, implicitly closing the change stream
-  if (change == null) return closeWithError(changeStream, CHANGESTREAM_CLOSED_ERROR, callback);
+  if (change == null) {
+    return closeWithError(changeStream, CHANGESTREAM_CLOSED_ERROR, callback);
+  }
 
-  if (change && !change._id) return closeWithError(changeStream, NO_RESUME_TOKEN_ERROR, callback);
+  if (change && !change._id) {
+    return closeWithError(changeStream, NO_RESUME_TOKEN_ERROR, callback);
+  }
 
   // cache the resume token
   changeStream.cursor?.cacheResumeToken(change._id);

src/cmap/connect.ts

@@ -277,7 +277,7 @@ function makeConnection(
       : typeof options.connectTimeoutMS === 'number'
       ? options.connectTimeoutMS
       : 30000;
-  const socketTimeout = typeof options.socketTimeout === 'number' ? options.socketTimeout : 360000;
+  const socketTimeout = typeof options.socketTimeout === 'number' ? options.socketTimeout : 0;
   const rejectUnauthorized =
     typeof options.rejectUnauthorized === 'boolean' ? options.rejectUnauthorized : true;
 

src/cmap/connection.ts

@@ -99,7 +99,7 @@ export class Connection extends EventEmitter {
     super(options);
     this.id = options.id;
     this.address = streamIdentifier(stream);
-    this.socketTimeout = options.socketTimeout ?? 360000;
+    this.socketTimeout = options.socketTimeout ?? 0;
     this.monitorCommands = options.monitorCommands ?? options.monitorCommands;
     this.closed = false;
     this.destroyed = false;

src/cursor/aggregation_cursor.ts

@@ -1,7 +1,6 @@
 import { MongoError } from '../error';
 import { Cursor, CursorOptions } from './cursor';
 import { CursorState } from './core_cursor';
-import { deprecate } from 'util';
 import type { AggregateOperation, AggregateOptions } from '../operations/aggregate';
 import type { Document } from '../bson';
 import type { Sort } from '../sort';
@@ -110,8 +109,8 @@ export class AggregationCursor extends Cursor<AggregateOperation, AggregationCur
 
   // deprecated methods
   /** @deprecated Add a geoNear stage to the aggregation pipeline */
-  geoNear = deprecate(($geoNear: Document) => {
+  geoNear($geoNear: Document): this {
     this.operation.addToPipeline({ $geoNear });
     return this;
-  }, 'The `$geoNear` stage is deprecated in MongoDB 4.0, and removed in version 4.2.');
+  }
 }

src/cursor/cursor.ts

@@ -126,6 +126,77 @@ export class CursorStream extends Readable {
   }
 }
 
+const kCursor = Symbol('cursor');
+
+export class CursorStream extends Readable {
+  [kCursor]: Cursor;
+  options: CursorStreamOptions;
+
+  /** @event */
+  static readonly CLOSE = 'close' as const;
+  /** @event */
+  static readonly DATA = 'data' as const;
+  /** @event */
+  static readonly END = 'end' as const;
+  /** @event */
+  static readonly FINISH = 'finish' as const;
+  /** @event */
+  static readonly ERROR = 'error' as const;
+  /** @event */
+  static readonly PAUSE = 'pause' as const;
+  /** @event */
+  static readonly READABLE = 'readable' as const;
+  /** @event */
+  static readonly RESUME = 'resume' as const;
+
+  constructor(cursor: Cursor, options?: CursorStreamOptions) {
+    super({ objectMode: true });
+    this[kCursor] = cursor;
+    this.options = options || {};
+  }
+
+  destroy(err?: AnyError): void {
+    this.pause();
+    this[kCursor].close();
+    super.destroy(err);
+  }
+
+  /** @internal */
+  _read(): void {
+    const cursor = this[kCursor];
+    if ((cursor.s && cursor.s.state === CursorState.CLOSED) || cursor.isDead()) {
+      this.push(null);
+      return;
+    }
+
+    // Get the next item
+    cursor._next((err, result) => {
+      if (err) {
+        if (cursor.s && cursor.s.state === CursorState.CLOSED) return;
+        if (!cursor.isDead()) this.emit(CursorStream.ERROR, err);
+        cursor.close(() => this.emit(CursorStream.END));
+        return;
+      }
+
+      // If we provided a transformation method
+      if (typeof this.options.transform === 'function' && result != null) {
+        this.push(this.options.transform(result));
+        return;
+      }
+
+      // Return the result
+      this.push(result);
+
+      if (result === null && cursor.isDead()) {
+        this.once(CursorStream.END, () => {
+          cursor.close();
+          this.emit(CursorStream.FINISH);
+        });
+      }
+    });
+  }
+}
+
 /**
  * **CURSORS Cannot directly be instantiated**
  * The `Cursor` class is an internal class that embodies a cursor on MongoDB

src/mongo_client.ts

@@ -17,6 +17,7 @@ import type { AuthMechanism } from './cmap/auth/defaultAuthProviders';
 import type { Topology } from './sdam/topology';
 import type { ClientSession, ClientSessionOptions } from './sessions';
 import type { OperationParent } from './operations/command';
+import type { TagSet } from './sdam/server_description';
 
 /** @public */
 export enum LogLevel {
@@ -97,7 +98,7 @@ export interface MongoURIOptions extends Pick<WriteConcernOptions, 'journal' | '
   /** Specifies, in seconds, how stale a secondary can be before the client stops using it for read operations. */
   maxStalenessSeconds?: number;
   /** Specifies the tags document as a comma-separated list of colon-separated key-value pairs.  */
-  readPreferenceTags?: string;
+  readPreferenceTags?: TagSet[];
   /** Specify the database name associated with the user’s credentials. */
   authSource?: string;
   /** Specify the authentication mechanism that MongoDB will use to authenticate the connection. */
@@ -220,6 +221,7 @@ export interface MongoClientPrivate {
   sessions: Set<ClientSession>;
   readConcern?: ReadConcern;
   writeConcern?: WriteConcern;
+  readPreference: ReadPreference;
   namespace: MongoDBNamespace;
   logger: Logger;
 }
@@ -281,6 +283,7 @@ export class MongoClient extends EventEmitter implements OperationParent {
       sessions: new Set(),
       readConcern: ReadConcern.fromOptions(options),
       writeConcern: WriteConcern.fromOptions(options),
+      readPreference: ReadPreference.fromOptions(options) || ReadPreference.primary,
       namespace: new MongoDBNamespace('admin'),
       logger: options?.logger ?? new Logger('MongoClient')
     };
@@ -295,7 +298,7 @@ export class MongoClient extends EventEmitter implements OperationParent {
   }
 
   get readPreference(): ReadPreference {
-    return ReadPreference.primary;
+    return this.s.readPreference;
   }
 
   get logger(): Logger {

src/operations/connect.ts

@@ -211,7 +211,7 @@ export function connect(
     const finalOptions = createUnifiedOptions(urlOptions, options);
 
     // Check if we have connection and socket timeout set
-    if (finalOptions.socketTimeoutMS == null) finalOptions.socketTimeoutMS = 360000;
+    if (finalOptions.socketTimeoutMS == null) finalOptions.socketTimeoutMS = 0;
     if (finalOptions.connectTimeoutMS == null) finalOptions.connectTimeoutMS = 10000;
     if (finalOptions.retryWrites == null) finalOptions.retryWrites = true;
     if (finalOptions.useRecoveryToken == null) finalOptions.useRecoveryToken = true;
@@ -582,7 +582,7 @@ function translateOptions(options: any) {
   }
 
   // Set the socket and connection timeouts
-  if (options.socketTimeoutMS == null) options.socketTimeoutMS = 360000;
+  if (options.socketTimeoutMS == null) options.socketTimeoutMS = 0;
   if (options.connectTimeoutMS == null) options.connectTimeoutMS = 10000;
 
   const translations = {

test/functional/change_stream.test.js

@@ -29,19 +29,16 @@ function withChangeStream(dbName, collectionName, callback) {
 
   return withClient((client, done) => {
     const db = client.db(dbName);
-    db.dropCollection(collectionName)
-      .catch(() => {})
-      .then(() =>
-        db.createCollection(collectionName, { w: 'majority' }, (err, collection) => {
-          if (err) return done(err);
-          withCursor(
-            collection.watch(),
-            (cursor, done) => callback(collection, cursor, done),
-            err => collection.drop(dropErr => done(err || dropErr))
-          );
-        })
-      )
-      .catch(done);
+    db.dropCollection(collectionName, () => {
+      db.createCollection(collectionName, { w: 'majority' }, (err, collection) => {
+        if (err) return done(err);
+        withCursor(
+          collection.watch(),
+          (cursor, done) => callback(collection, cursor, done),
+          err => collection.drop(dropErr => done(err || dropErr))
+        );
+      });
+    });
   });
 }
 
@@ -66,7 +63,18 @@ function triggerResumableError(changeStream, delay, onClose) {
   });
 
   function triggerError() {
-    changeStream.triggerError(new MongoNetworkError('fake error'));
+    const cursorStream = changeStream.cursorStream;
+    if (cursorStream) {
+      cursorStream.emit('error', new MongoNetworkError('error triggered from test'));
+      return;
+    }
+
+    const nextStub = sinon.stub(changeStream.cursor, 'next').callsFake(function (callback) {
+      callback(new MongoNetworkError('error triggered from test'));
+      nextStub.restore();
+    });
+
+    changeStream.next(() => {});
   }
 
   if (delay != null) {

test/functional/mongo_client.test.js

@@ -3,6 +3,7 @@
 var f = require('util').format;
 var test = require('./shared').assert;
 var setupDatabase = require('./shared').setupDatabase;
+const { ReadPreference } = require('../../src');
 const { Db } = require('../../src/db');
 const expect = require('chai').expect;
 
@@ -301,4 +302,10 @@ describe('MongoClient', function () {
         }
       });
   });
+
+  it('should cache a resolved readPreference from options', function () {
+    const client = this.configuration.newClient({}, { readPreference: ReadPreference.SECONDARY });
+    expect(client.readPreference).to.be.instanceOf(ReadPreference);
+    expect(client.readPreference).to.have.property('mode', ReadPreference.SECONDARY);
+  });
 });