fix(changeStream): properly handle changeStream event mid-close

lib/change_stream.js

@@ -368,6 +368,15 @@ function processNewChange(args) {
   const change = args.change;
   const callback = args.callback;
   const eventEmitter = args.eventEmitter || false;
+
+  if (changeStream.isClosed()) {
+    if (eventEmitter) return;
+
+    const error = new Error('ChangeStream is closed');
+    if (typeof callback === 'function') return callback(error, null);
+    return changeStream.promiseLibrary.reject(error);
+  }
+
   const topology = changeStream.topology;
   const options = changeStream.cursor.options;
 

test/functional/change_stream_tests.js

@@ -1677,4 +1677,133 @@ describe('Change Streams', function() {
         .then(() => finish(), err => finish(err));
     }
   });
+
+  describe('should properly handle a changeStream event being processed mid-close', function() {
+    it('when invoked with promises', {
+      metadata: { requires: { topology: 'replicaset', mongodb: '>=3.5.10' } },
+      test: function() {
+        function read(coll) {
+          const changeStream = coll.watch();
+          return Promise.resolve()
+            .then(() => changeStream.next())
+            .then(() => changeStream.next())
+            .then(() => {
+              const closeP = Promise.resolve().then(() => changeStream.close());
+              const nextP = changeStream.next();
+
+              return closeP.then(() => nextP);
+            });
+        }
+
+        function write(coll) {
+          return Promise.resolve()
+            .then(() => coll.insertOne({ a: 1 }))
+            .then(() => coll.insertOne({ b: 2 }))
+            .then(() => coll.insertOne({ c: 3 }));
+        }
+
+        const client = this.configuration.newClient();
+
+        return client.connect().then(() => {
+          const coll = client.db(this.configuration.db).collection('tester');
+
+          return Promise.all([read(coll), write(coll)])
+            .then(
+              () => Promise.reject(new Error('Expected operation to fail with error')),
+              err => expect(err.message).to.equal('ChangeStream is closed')
+            )
+            .then(() => client.close(), err => client.close().then(() => Promise.reject(err)));
+        });
+      }
+    });
+
+    it('when invoked with callbacks', {
+      metadata: { requires: { topology: 'replicaset', mongodb: '>=3.5.10' } },
+      test: function(done) {
+        const client = this.configuration.newClient();
+        let closed = false;
+        const close = err => {
+          if (closed) {
+            return;
+          }
+          closed = true;
+          return client.close(() => done(err));
+        };
+        client.connect(err => {
+          if (err) {
+            return close(err);
+          }
+
+          const coll = client.db(this.configuration.db).collection('tester');
+          const changeStream = coll.watch();
+
+          changeStream.next(() => {
+            changeStream.next(() => {
+              changeStream.next(err => {
+                let _err = null;
+                try {
+                  expect(err.message).to.equal('ChangeStream is closed');
+                } catch (e) {
+                  _err = e;
+                } finally {
+                  close(_err);
+                }
+              });
+              changeStream.close();
+            });
+          });
+
+          coll.insertOne({ a: 1 }, () => {
+            coll.insertOne({ b: 2 }, () => {
+              coll.insertOne({ c: 3 }, () => {});
+            });
+          });
+        });
+      }
+    });
+
+    it('when invoked using eventEmitter API', {
+      metadata: { requires: { topology: 'replicaset', mongodb: '>=3.5.10' } },
+      test: function(done) {
+        const client = this.configuration.newClient();
+        let closed = false;
+        const close = _err => {
+          if (closed) {
+            return;
+          }
+          closed = true;
+          return client.close(() => done(_err));
+        };
+
+        client.connect(err => {
+          if (err) {
+            return close(err);
+          }
+
+          const coll = client.db(this.configuration.db).collection('tester');
+          const changeStream = coll.watch();
+
+          let counter = 0;
+          changeStream.on('change', () => {
+            counter += 1;
+            if (counter === 2) {
+              changeStream.close();
+              setTimeout(() => close());
+            } else if (counter >= 3) {
+              close(new Error('Should not have received more than 2 events'));
+            }
+          });
+          changeStream.on('error', err => close(err));
+
+          setTimeout(() => {
+            coll.insertOne({ a: 1 }, () => {
+              coll.insertOne({ b: 2 }, () => {
+                coll.insertOne({ c: 3 }, () => {});
+              });
+            });
+          });
+        });
+      }
+    });
+  });
 });