refactor(ChangeStream): use maybePromise for next/hasNext

NODE-2570

Author: emadum

lib/change_stream.js

@@ -4,6 +4,7 @@ const EventEmitter = require('events');
 const { MongoError, isResumableError } = require('./error');
 const { Cursor } = require('./cursor');
 const { relayEvents, maxWireVersion } = require('./utils');
+const maybePromise = require('./utils').maybePromise;
 const AggregateOperation = require('./operations/aggregate');
 
 const CHANGE_STREAM_OPTIONS = ['resumeAfter', 'startAfter', 'startAtOperationTime', 'fullDocument'];
@@ -124,10 +125,10 @@ class ChangeStream extends EventEmitter {
    * @function ChangeStream.prototype.hasNext
    * @param {ChangeStream~resultCallback} [callback] The result callback.
    * @throws {MongoError}
-   * @returns {Promise} returns Promise if no callback passed
+   * @returns {Promise|void} returns Promise if no callback passed
    */
   hasNext(callback) {
-    return this.cursor.hasNext(callback);
+    return maybePromise(this.parent, callback, cb => this.cursor.hasNext(cb));
   }
 
   /**
@@ -136,19 +137,17 @@ class ChangeStream extends EventEmitter {
    * @function ChangeStream.prototype.next
    * @param {ChangeStream~resultCallback} [callback] The result callback.
    * @throws {MongoError}
-   * @returns {Promise} returns Promise if no callback passed
+   * @returns {Promise|void} returns Promise if no callback passed
    */
   next(callback) {
-    var self = this;
-    if (this.isClosed()) {
-      if (callback) return callback(new Error('Change Stream is not open.'), null);
-      return self.promiseLibrary.reject(new Error('Change Stream is not open.'));
-    }
-
-    return this.cursor
-      .next()
-      .then(change => processNewChange({ changeStream: self, change, callback }))
-      .catch(error => processNewChange({ changeStream: self, error, callback }));
+    return maybePromise(this.parent, callback, cb => {
+      if (this.isClosed()) {
+        return cb(new Error('Change Stream is not open.'));
+      }
+      this.cursor.next((error, change) => {
+        processNewChange({ changeStream: this, error, change, callback: cb });
+      });
+    });
   }
 
   /**