refactor(ChangeStream): use maybePromise for close, improve tests

NODE-2598

Author: emadum

lib/change_stream.js

@@ -93,6 +93,8 @@ class ChangeStream extends EventEmitter {
     // Create contained Change Stream cursor
     this.cursor = createChangeStreamCursor(this, options);
 
+    this.closed = false;
+
     // Listen for any `change` listeners being added to ChangeStream
     this.on('newListener', eventName => {
       if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {
@@ -142,7 +144,7 @@ class ChangeStream extends EventEmitter {
   next(callback) {
     return maybePromise(this.parent, callback, cb => {
       if (this.isClosed()) {
-        return cb(new Error('Change Stream is not open.'));
+        return cb(new MongoError('ChangeStream is closed'));
       }
       this.cursor.next((error, change) => {
         processNewChange({ changeStream: this, error, change, callback: cb });
@@ -157,10 +159,7 @@ class ChangeStream extends EventEmitter {
    * @returns {boolean}
    */
   isClosed() {
-    if (this.cursor) {
-      return this.cursor.isClosed();
-    }
-    return true;
+    return this.closed || (this.cursor && this.cursor.isClosed());
   }
 
   /**
@@ -171,31 +170,20 @@ class ChangeStream extends EventEmitter {
    * @returns {Promise} returns Promise if no callback passed
    */
   close(callback) {
-    if (!this.cursor) {
-      if (callback) return callback();
-      return this.promiseLibrary.resolve();
-    }
+    return maybePromise(this.parent, callback, cb => {
+      if (this.closed) return cb();
 
-    // Tidy up the existing cursor
-    const cursor = this.cursor;
+      // flag the change stream as explicitly closed
+      this.closed = true;
 
-    if (callback) {
-      return cursor.close(err => {
-        ['data', 'close', 'end', 'error'].forEach(event => cursor.removeAllListeners(event));
-        delete this.cursor;
+      // Tidy up the existing cursor
+      const cursor = this.cursor;
 
-        return callback(err);
-      });
-    }
-
-    const PromiseCtor = this.promiseLibrary || Promise;
-    return new PromiseCtor((resolve, reject) => {
-      cursor.close(err => {
+      return cursor.close(err => {
         ['data', 'close', 'end', 'error'].forEach(event => cursor.removeAllListeners(event));
-        delete this.cursor;
+        this.cursor = undefined;
 
-        if (err) return reject(err);
-        resolve();
+        return cb(err);
       });
     });
   }
@@ -329,7 +317,7 @@ class ChangeStreamCursor extends Cursor {
   _initializeCursor(callback) {
     super._initializeCursor((err, result) => {
       if (err) {
-        callback(err, null);
+        callback(err);
         return;
       }
 
@@ -355,7 +343,7 @@ class ChangeStreamCursor extends Cursor {
   _getMore(callback) {
     super._getMore((err, response) => {
       if (err) {
-        callback(err, null);
+        callback(err);
         return;
       }
 
@@ -467,12 +455,12 @@ function waitForTopologyConnected(topology, options, callback) {
     const timeout = options.timeout || SELECTION_TIMEOUT;
     const readPreference = options.readPreference;
 
-    if (topology.isConnected({ readPreference })) return callback(null, null);
+    if (topology.isConnected({ readPreference })) return callback();
     const hrElapsed = process.hrtime(start);
     const elapsed = (hrElapsed[0] * 1e9 + hrElapsed[1]) / 1e6;
     if (elapsed > timeout) return callback(new MongoError('Timed out waiting for connection'));
     waitForTopologyConnected(topology, options, callback);
-  }, 3000); // this is an arbitrary wait time to allow SDAM to transition
+  }, 500); // this is an arbitrary wait time to allow SDAM to transition
 }
 
 // Handle new change events. This method brings together the routes from the callback, event emitter, and promise ways of using ChangeStream.
@@ -484,17 +472,15 @@ function processNewChange(args) {
   const eventEmitter = args.eventEmitter || false;
   const cursor = changeStream.cursor;
 
-  // If the cursor is null, then it should not process a change.
-  if (cursor == null) {
+  // If the cursor is null or the change stream has been closed explictly, do not process a change.
+  if (cursor == null || changeStream.closed) {
     // We do not error in the eventEmitter case.
+    changeStream.closed = true;
     if (eventEmitter) {
       return;
     }
-
-    const error = new MongoError('ChangeStream is closed');
-    return typeof callback === 'function'
-      ? callback(error, null)
-      : changeStream.promiseLibrary.reject(error);
+    callback(new MongoError('ChangeStream is closed'));
+    return;
   }
 
   const topology = changeStream.topology;
@@ -513,46 +499,27 @@ function processNewChange(args) {
       // close internal cursor, ignore errors
       changeStream.cursor.close();
 
-      // attempt recreating the cursor
-      if (eventEmitter) {
-        waitForTopologyConnected(topology, { readPreference: options.readPreference }, err => {
-          if (err) {
+      waitForTopologyConnected(topology, { readPreference: options.readPreference }, err => {
+        if (err) {
+          // if there's an error reconnecting, close the change stream
+          changeStream.closed = true;
+          if (eventEmitter) {
             changeStream.emit('error', err);
             changeStream.emit('close');
             return;
           }
-          changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions);
-        });
+          return callback(err);
+        }
 
-        return;
-      }
-
-      if (callback) {
-        waitForTopologyConnected(topology, { readPreference: options.readPreference }, err => {
-          if (err) return callback(err, null);
-
-          changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions);
-          changeStream.next(callback);
-        });
-
-        return;
-      }
-
-      return new Promise((resolve, reject) => {
-        waitForTopologyConnected(topology, { readPreference: options.readPreference }, err => {
-          if (err) return reject(err);
-          resolve();
-        });
-      })
-        .then(
-          () => (changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions))
-        )
-        .then(() => changeStream.next());
+        changeStream.cursor = createChangeStreamCursor(changeStream, cursor.resumeOptions);
+        if (eventEmitter) return;
+        changeStream.next(callback);
+      });
+      return;
     }
 
     if (eventEmitter) return changeStream.emit('error', error);
-    if (typeof callback === 'function') return callback(error, null);
-    return changeStream.promiseLibrary.reject(error);
+    return callback(error);
   }
 
   changeStream.attemptingResume = false;
@@ -563,8 +530,7 @@ function processNewChange(args) {
     );
 
     if (eventEmitter) return changeStream.emit('error', noResumeTokenError);
-    if (typeof callback === 'function') return callback(noResumeTokenError, null);
-    return changeStream.promiseLibrary.reject(noResumeTokenError);
+    return callback(noResumeTokenError);
   }
 
   // cache the resume token
@@ -576,8 +542,7 @@ function processNewChange(args) {
 
   // Return the change
   if (eventEmitter) return changeStream.emit('change', change);
-  if (typeof callback === 'function') return callback(error, change);
-  return changeStream.promiseLibrary.resolve(change);
+  return callback(error, change);
 }
 
 /**