throw error if calling methods on change stream that errored out

Author: vkarpov15

lib/cursor/changeStream.js

@@ -5,6 +5,7 @@
  */
 
 const EventEmitter = require('events').EventEmitter;
+const MongooseError = require('../error/mongooseError');
 
 /*!
  * ignore
@@ -25,6 +26,7 @@ class ChangeStream extends EventEmitter {
     this.bindedEvents = false;
     this.pipeline = pipeline;
     this.options = options;
+    this.errored = false;
 
     if (options && options.hydrate && !options.model) {
       throw new Error(
@@ -39,6 +41,7 @@ class ChangeStream extends EventEmitter {
       try {
         changeStreamThunk((err, driverChangeStream) => {
           if (err != null) {
+            this.errored = true;
             this.emit('error', err);
             return reject(err);
           }
@@ -49,6 +52,7 @@ class ChangeStream extends EventEmitter {
         });
       } catch (err) {
         syncError = err;
+        this.errored = true;
         this.emit('error', err);
         reject(err);
       }
@@ -107,10 +111,16 @@ class ChangeStream extends EventEmitter {
   }
 
   hasNext(cb) {
+    if (this.errored) {
+      throw new MongooseError('Cannot call hasNext() on errored ChangeStream');
+    }
     return this.driverChangeStream.hasNext(cb);
   }
 
   next(cb) {
+    if (this.errored) {
+      throw new MongooseError('Cannot call next() on errored ChangeStream');
+    }
     if (this.options && this.options.hydrate) {
       if (cb != null) {
         const originalCb = cb;
@@ -141,16 +151,25 @@ class ChangeStream extends EventEmitter {
   }
 
   addListener(event, handler) {
+    if (this.errored) {
+      throw new MongooseError('Cannot call addListener() on errored ChangeStream');
+    }
     this._bindEvents();
     return super.addListener(event, handler);
   }
 
   on(event, handler) {
+    if (this.errored) {
+      throw new MongooseError('Cannot call on() on errored ChangeStream');
+    }
     this._bindEvents();
     return super.on(event, handler);
   }
 
   once(event, handler) {
+    if (this.errored) {
+      throw new MongooseError('Cannot call once() on errored ChangeStream');
+    }
     this._bindEvents();
     return super.once(event, handler);
   }