feat(ChangeStreamCursor): introduce new cursor type for change streams

This takes the approach of encapsulating all change stream related
behavior into a new `ChangeStreamCursor` class (rather than the
alternate approach of using a `ResumeTokenTracker`). Notably, this
reintroduces `_initializeCursor` as a publicly accessible, but
private method on the base `Cursor` class, allowing us to hook into
initial command execution as well as subsequent `getMore`.

Author: mbroadst

lib/change_stream.js

@@ -5,97 +5,20 @@ const isResumableError = require('./error').isResumableError;
 const MongoError = require('./core').MongoError;
 const ReadConcern = require('./read_concern');
 const MongoDBNamespace = require('./utils').MongoDBNamespace;
+const Cursor = require('./cursor');
+const relayEvents = require('./core/utils').relayEvents;
+const maxWireVersion = require('./core/utils').maxWireVersion;
 
-var cursorOptionNames = ['maxAwaitTimeMS', 'collation', 'readPreference'];
+const CHANGE_STREAM_OPTIONS = ['resumeAfter', 'startAfter', 'startAtOperationTime', 'fullDocument'];
+const CURSOR_OPTIONS = ['batchSize', 'maxAwaitTimeMS', 'collation', 'readPreference'].concat(
+  CHANGE_STREAM_OPTIONS
+);
 
 const CHANGE_DOMAIN_TYPES = {
   COLLECTION: Symbol('Collection'),
   DATABASE: Symbol('Database'),
   CLUSTER: Symbol('Cluster')
 };
-class ResumeTokenTracker extends EventEmitter {
-  constructor(changeStream, options) {
-    super();
-    this.changeStream = changeStream;
-    this.options = options;
-    this._postBatchResumeToken = undefined;
-  }
-
-  set resumeToken(token) {
-    this._resumeToken = token;
-    // NOTE: Event is for internal use only, and is not part of public API
-    this.emit('tokenChange');
-  }
-
-  get resumeToken() {
-    return this._resumeToken;
-  }
-
-  init() {
-    this._resumeToken = this.options.startAfter || this.options.resumeAfter;
-    this._operationTime = this.options.startAtOperationTime;
-    this._init = true;
-  }
-
-  resumeInfo() {
-    const resumeInfo = {};
-
-    if (this._init && this.resumeToken) {
-      resumeInfo.resumeAfter = this.resumeToken;
-    } else if (this._init && this._operationTime) {
-      resumeInfo.startAtOperationTime = this._operationTime;
-    } else {
-      if (this.options.startAfter) {
-        resumeInfo.startAfter = this.options.startAfter;
-      }
-
-      if (this.options.resumeAfter) {
-        resumeInfo.resumeAfter = this.options.resumeAfter;
-      }
-
-      if (this.options.startAtOperationTime) {
-        resumeInfo.startAtOperationTime = this.options.startAtOperationTime;
-      }
-    }
-
-    return resumeInfo;
-  }
-
-  onResponse(postBatchResumeToken, operationTime) {
-    if (this.changeStream.isClosed()) {
-      return;
-    }
-    const cursor = this.changeStream.cursor;
-    if (!postBatchResumeToken) {
-      if (
-        !(this.resumeToken || this._operationTime || cursor.bufferedCount()) &&
-        cursor.server &&
-        cursor.server.ismaster.maxWireVersion >= 7
-      ) {
-        this._operationTime = operationTime;
-      }
-    } else {
-      this._postBatchResumeToken = postBatchResumeToken;
-      if (cursor.cursorState.documents.length === 0) {
-        this.resumeToken = this._postBatchResumeToken;
-      }
-    }
-
-    // NOTE: Event is for internal use only, and is not part of public API
-    this.emit('response');
-  }
-
-  onNext(doc) {
-    if (this.changeStream.isClosed()) {
-      return;
-    }
-    if (this._postBatchResumeToken && this.changeStream.cursor.bufferedCount() === 0) {
-      this.resumeToken = this._postBatchResumeToken;
-    } else {
-      this.resumeToken = doc._id;
-    }
-  }
-}
 
 /**
  * @typedef ResumeToken
@@ -133,6 +56,7 @@ class ResumeTokenTracker extends EventEmitter {
  * @fires ChangeStream#change
  * @fires ChangeStream#end
  * @fires ChangeStream#error
+ * @fires ChangeStream#resumeTokenChanged
  * @return {ChangeStream} a ChangeStream instance.
  */
 class ChangeStream extends EventEmitter {
@@ -170,12 +94,8 @@ class ChangeStream extends EventEmitter {
       this.options.readPreference = changeDomain.s.readPreference;
     }
 
-    this._resumeTokenTracker = new ResumeTokenTracker(this, options);
-
     // Create contained Change Stream cursor
-    this.cursor = createChangeStreamCursor(this);
-
-    this._resumeTokenTracker.init();
+    this.cursor = createChangeStreamCursor(this, options);
 
     // Listen for any `change` listeners being added to ChangeStream
     this.on('newListener', eventName => {
@@ -200,7 +120,7 @@ class ChangeStream extends EventEmitter {
    * after the most recently returned change.
    */
   get resumeToken() {
-    return this._resumeTokenTracker.resumeToken;
+    return this.cursor.resumeToken;
   }
 
   /**
@@ -321,9 +241,93 @@ class ChangeStream extends EventEmitter {
   }
 }
 
+class ChangeStreamCursor extends Cursor {
+  constructor(bson, ns, cmd, options, topology, topologyOptions) {
+    // TODO: spread will help a lot here
+    super(bson, ns, cmd, options, topology, topologyOptions);
+
+    options = options || {};
+    this._resumeToken = null;
+    this.startAtOperationTime = options.startAtOperationTime;
+
+    if (options.startAfter) {
+      this.resumeToken = options.startAfter;
+    } else if (options.resumeAfter) {
+      this.resumeToken = options.resumeAfter;
+    }
+  }
+
+  set resumeToken(token) {
+    this._resumeToken = token;
+    this.emit('resumeTokenChanged', token);
+  }
+
+  get resumeToken() {
+    return this._resumeToken;
+  }
+
+  get resumeOptions() {
+    if (this.resumeToken) {
+      return { resumeAfter: this.resumeToken };
+    }
+
+    if (this.startAtOperationTime && maxWireVersion(this.server) >= 7) {
+      return { startAtOperationTime: this.startAtOperationTime };
+    }
+
+    return null;
+  }
+
+  _initializeCursor(callback) {
+    super._initializeCursor((err, result) => {
+      if (err) {
+        callback(err, null);
+        return;
+      }
+
+      const response = result.documents[0];
+
+      if (
+        this.startAtOperationTime == null &&
+        this.resumeAfter == null &&
+        this.startAfter == null &&
+        maxWireVersion(this.server) >= 7
+      ) {
+        this.startAtOperationTime = response.operationTime;
+      }
+
+      const cursor = response.cursor;
+      if (cursor.firstBatch.length === 0 && cursor.postBatchResumeToken) {
+        this.resumeToken = cursor.postBatchResumeToken;
+      }
+
+      this.emit('response');
+      callback(err, result);
+    });
+  }
+
+  _getMore(callback) {
+    super._getMore((err, response) => {
+      if (err) {
+        callback(err, null);
+        return;
+      }
+
+      const cursor = response.cursor;
+      if (cursor.nextBatch.length === 0 && cursor.postBatchResumeToken) {
+        this.resumeToken = cursor.postBatchResumeToken;
+      }
+
+      this.emit('response');
+      callback(err, response);
+    });
+  }
+}
+
 // Create a new change stream cursor based on self's configuration
-var createChangeStreamCursor = function(self) {
-  var changeStreamCursor = buildChangeStreamAggregationCommand(self);
+var createChangeStreamCursor = function(self, options) {
+  const changeStreamCursor = buildChangeStreamAggregationCommand(self, options);
+  relayEvents(changeStreamCursor, self, ['resumeTokenChanged', 'end', 'close']);
 
   /**
    * Fired for each new matching change in the specified namespace. Attaching a `change`
@@ -345,19 +349,13 @@ var createChangeStreamCursor = function(self) {
    * @event ChangeStream#close
    * @type {null}
    */
-  changeStreamCursor.on('close', function() {
-    self.emit('close');
-  });
 
   /**
    * Change stream end event
    *
    * @event ChangeStream#end
    * @type {null}
    */
-  changeStreamCursor.on('end', function() {
-    self.emit('end');
-  });
 
   /**
    * Fired when the stream encounters an error.
@@ -379,25 +377,26 @@ var createChangeStreamCursor = function(self) {
   return changeStreamCursor;
 };
 
-var buildChangeStreamAggregationCommand = function(self) {
+function applyKnownOptions(target, source, optionNames) {
+  optionNames.forEach(name => {
+    if (source[name]) {
+      target[name] = source[name];
+    }
+  });
+}
+
+var buildChangeStreamAggregationCommand = function(self, options) {
+  options = options || {};
   const topology = self.topology;
   const namespace = self.namespace;
   const pipeline = self.pipeline;
-  const options = self.options;
-  const resumeTokenTracker = self._resumeTokenTracker;
 
-  const changeStreamStageOptions = Object.assign(
-    { fullDocument: options.fullDocument || 'default' },
-    resumeTokenTracker.resumeInfo()
-  );
+  const changeStreamStageOptions = { fullDocument: options.fullDocument || 'default' };
+  applyKnownOptions(changeStreamStageOptions, options, CHANGE_STREAM_OPTIONS);
 
   // Map cursor options
-  var cursorOptions = { resumeTokenTracker };
-  cursorOptionNames.forEach(function(optionName) {
-    if (options[optionName]) {
-      cursorOptions[optionName] = options[optionName];
-    }
-  });
+  const cursorOptions = { cursorFactory: ChangeStreamCursor };
+  applyKnownOptions(cursorOptions, options, CURSOR_OPTIONS);
 
   if (self.type === CHANGE_DOMAIN_TYPES.CLUSTER) {
     changeStreamStageOptions.allChangesForCluster = true;
@@ -460,6 +459,7 @@ function processNewChange(args) {
       : changeStream.promiseLibrary.reject(error);
   }
 
+  const cursor = changeStream.cursor;
   const topology = changeStream.topology;
   const options = changeStream.cursor.options;
 
@@ -483,7 +483,7 @@ function processNewChange(args) {
             changeStream.emit('close');
             return;
           }
-          changeStream.cursor = createChangeStreamCursor(changeStream);
+          changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions);
         });
 
         return;
@@ -493,7 +493,7 @@ function processNewChange(args) {
         waitForTopologyConnected(topology, { readPreference: options.readPreference }, err => {
           if (err) return callback(err, null);
 
-          changeStream.cursor = createChangeStreamCursor(changeStream);
+          changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions);
           changeStream.next(callback);
         });
 
@@ -506,7 +506,9 @@ function processNewChange(args) {
           resolve();
         });
       })
-        .then(() => (changeStream.cursor = createChangeStreamCursor(changeStream)))
+        .then(
+          () => (changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions))
+        )
         .then(() => changeStream.next());
     }
 
@@ -517,9 +519,8 @@ function processNewChange(args) {
 
   changeStream.attemptingResume = false;
 
-  // Cache the resume token if it is present. If it is not present return an error.
-  if (!change || !change._id) {
-    var noResumeTokenError = new Error(
+  if (change && !change._id) {
+    const noResumeTokenError = new Error(
       'A change stream document has been received that lacks a resume token (_id).'
     );
 
@@ -528,7 +529,12 @@ function processNewChange(args) {
     return changeStream.promiseLibrary.reject(noResumeTokenError);
   }
 
-  changeStream._resumeTokenTracker.onNext(change);
+  // cache the resume token
+  if (cursor.bufferedCount() === 0 && cursor.cursorState.postBatchResumeToken) {
+    cursor.resumeToken = cursor.cursorState.postBatchResumeToken;
+  } else {
+    cursor.resumeToken = change._id;
+  }
 
   // wipe the startAtOperationTime if there was one so that there won't be a conflict
   // between resumeToken and startAtOperationTime if we need to reconnect the cursor

lib/command_cursor.js

@@ -150,7 +150,8 @@ var methodsToInherit = [
   'kill',
   'setCursorBatchSize',
   '_find',
-  '_getmore',
+  '_initializeCursor',
+  '_getMore',
   '_killcursor',
   'isDead',
   'explain',