Merge pull request #3474 from bigcup/fixChangeStreamIssue#1569

Removed observers from a Model after finish of the stream

lib/persisted-model.js

@@ -1851,36 +1851,42 @@ module.exports = function(registry) {
     var idName = this.getIdName();
     var Model = this;
     var changes = new PassThrough({objectMode: true});
-    var writeable = true;
 
-    changes.destroy = function() {
+    changes._destroy = function() {
-      changes.removeAllListeners('error');
+      changes.end();
-      changes.removeAllListeners('end');
+      changes.emit('end');
-      writeable = false;
+      changes.emit('close');
-      changes = null;
     };
 
-    changes.on('error', function() {
+    changes.destroy = changes.destroy || changes._destroy; // node 8 compability
-      writeable = false;
+
-    });
+    changes.on('error', removeHandlers);
-    changes.on('end', function() {
+    changes.on('close', removeHandlers);
-      writeable = false;
+    changes.on('finish', removeHandlers);
-    });
+    changes.on('end', removeHandlers);
 
     process.nextTick(function() {
       cb(null, changes);
     });
 
-    Model.observe('after save', createChangeHandler('save'));
+    Model.observe('after save', changeHandler);
-    Model.observe('after delete', createChangeHandler('delete'));
+    Model.observe('after delete', deleteHandler);
 
-    function createChangeHandler(type) {
+    function changeHandler(ctx, next) {
-      return function(ctx, next) {
+      var change = createChangeObject(ctx, 'save');
-        // since it might have set to null via destroy
+      changes.write(change);
-        if (!changes) {
-          return next();
-        }
 
+      next();
+    };
+
+    function deleteHandler(ctx, next) {
+      var change = createChangeObject(ctx, 'delete');
+      changes.write(change);
+
+      next();
+    };
+
+    function createChangeObject(ctx, type) {
       var where = ctx.where;
       var data = ctx.instance || ctx.data;
       var whereId = where && where[idName];
@@ -1917,13 +1923,12 @@ module.exports = function(registry) {
           break;
       }
 
-        // TODO(ritch) this is ugly... maybe a ReadableStream would be better
+      return change;
-        if (writeable) {
-          changes.write(change);
     }
 
-        next();
+    function removeHandlers() {
-      };
+      Model.removeObserver('after save', changeHandler);
+      Model.removeObserver('after delete', deleteHandler);
     }
   };
 

test/change-stream.test.js

@@ -5,6 +5,7 @@
 
 'use strict';
 var expect = require('./helpers/expect');
+var sinon = require('sinon');
 var loopback = require('../');
 
 describe('PersistedModel.createChangeStream()', function() {
@@ -20,6 +21,8 @@ describe('PersistedModel.createChangeStream()', function() {
       });
     });
 
+    afterEach(verifyObserversRemoval);
+
     it('should detect create', function(done) {
       var Score = this.Score;
 
@@ -27,7 +30,6 @@ describe('PersistedModel.createChangeStream()', function() {
         changes.on('data', function(change) {
           expect(change.type).to.equal('create');
           changes.destroy();
-
           done();
         });
 
@@ -67,6 +69,32 @@ describe('PersistedModel.createChangeStream()', function() {
         });
       });
     });
+
+    it('should not emit changes after destroy', function(done) {
+      var Score = this.Score;
+
+      var spy = sinon.spy();
+
+      Score.createChangeStream(function(err, changes) {
+        changes.on('data', function() {
+          spy();
+          changes.destroy();
+        });
+
+        Score.create({team: 'foo'})
+          .then(() => Score.deleteAll())
+          .then(() => {
+            expect(spy.calledOnce);
+            done();
+          });
+      });
+    });
+
+    function verifyObserversRemoval() {
+      var Score = this.Score;
+      expect(Score._observers['after save']).to.be.empty();
+      expect(Score._observers['after delete']).to.be.empty();
+    }
   });
 
   // TODO(ritch) implement multi-server support