(#2555) - fix 'change' event for db.sync()

lib/replicate.js

@@ -39,7 +39,6 @@ Replication.prototype.ready = function (src, target) {
   src.once('destroyed', onDestroy);
   target.once('destroyed', onDestroy);
   function cleanup() {
-    self.removeAllListeners();
     src.removeListener('destroyed', onDestroy);
     target.removeListener('destroyed', onDestroy);
   }

lib/sync.js

@@ -55,17 +55,25 @@ function Sync(src, target, opts, callback) {
   }
   var listeners = {};
 
-  function removal(event, func) {
-    if (event === 'change' &&
-      (func === pullChange ||
-      func === pushChange)) {
-      self.removeAllListeners('change');
-    } else if (event === 'cancel' &&
-      func === onCancel) {
-      self.removeAllListeners('cancel');
-    } else if (event in listeners && func === listeners[event]) {
-      self.removeAllListeners(event);
-    }
+  var removed = {};
+  function removeAll(type) { // type is 'push' or 'pull'
+    return function (event, func) {
+      var isChange = event === 'change' &&
+        (func === pullChange || func === pushChange);
+      var isCancel = event === 'cancel' && func === onCancel;
+      var isOtherEvent = event in listeners && func === listeners[event];
+
+      if (isChange || isCancel || isOtherEvent) {
+        if (!(event in removed)) {
+          removed[event] = {};
+        }
+        removed[event][type] = true;
+        if (Object.keys(removed[event]).length === 2) {
+          // both push and pull have asked to be removed
+          self.removeAllListeners(event);
+        }
+      }
+    };
   }
 
   this.on('newListener', function (event) {
@@ -76,6 +84,7 @@ function Sync(src, target, opts, callback) {
       self.pull.on('cancel', onCancel);
       self.push.on('cancel', onCancel);
     } else if (event !== 'error' &&
+      event !== 'removeListener' &&
       event !== 'complete' && !(event in listeners)) {
       listeners[event] = function (e) {
         self.emit(event, e);
@@ -101,8 +110,8 @@ function Sync(src, target, opts, callback) {
     }
   });
 
-  this.pull.on('removeListener', removal);
-  this.push.on('removeListener', removal);
+  this.pull.on('removeListener', removeAll('pull'));
+  this.push.on('removeListener', removeAll('push'));
 
   var promise = utils.Promise.all([
     this.push,

tests/test.sync.js

@@ -248,5 +248,88 @@ adapters.forEach(function (adapters) {
       });
       db.put(doc1);
     });
+
+    it('Push and pull changes both fire (issue 2555)', function (done) {
+      var db = new PouchDB(dbs.name);
+      var remote = new PouchDB(dbs.remote);
+
+      db.post({}).then(function () {
+        return remote.post({});
+      }).then(function () {
+        var numChanges = 0;
+        var lastChange;
+        var sync = db.sync(remote);
+        sync.on('change', function (change) {
+          ['push', 'pull'].should.contain(change.direction);
+          change.change.docs_read.should.equal(1);
+          change.change.docs_written.should.equal(1);
+          if (!lastChange) {
+            lastChange = change.direction;
+          } else {
+            lastChange.should.not.equal(change.direction);
+          }
+          if (++numChanges === 2) {
+            done();
+          }
+        });
+      });
+    });
+
+    it('Doesn\'t have a memory leak (push)', function (done) {
+      var db = new PouchDB(dbs.name);
+      var remote = new PouchDB(dbs.remote);
+
+      db.bulkDocs([{}, {}, {}]).then(function () {
+        return remote.bulkDocs([{}, {}, {}]);
+      }).then(function () {
+        var sync = db.replicate.to(remote);
+        sync.on('change', function () {});
+        sync.on('error', function () {});
+        sync.on('complete', function () {
+          setTimeout(function () {
+            Object.keys(sync._events).should.have.length(0);
+            done();
+          });
+        });
+      });
+    });
+
+    it('Doesn\'t have a memory leak (pull)', function (done) {
+      var db = new PouchDB(dbs.name);
+      var remote = new PouchDB(dbs.remote);
+
+      db.bulkDocs([{}, {}, {}]).then(function () {
+        return remote.bulkDocs([{}, {}, {}]);
+      }).then(function () {
+        var sync = db.replicate.from(remote);
+        sync.on('change', function () {});
+        sync.on('error', function () {});
+        sync.on('complete', function () {
+          setTimeout(function () {
+            Object.keys(sync._events).should.have.length(0);
+            done();
+          });
+        });
+      });
+    });
+
+    it('Doesn\'t have a memory leak (bi)', function (done) {
+      var db = new PouchDB(dbs.name);
+      var remote = new PouchDB(dbs.remote);
+
+      db.bulkDocs([{}, {}, {}]).then(function () {
+        return remote.bulkDocs([{}, {}, {}]);
+      }).then(function () {
+        var sync = db.sync(remote);
+        sync.on('change', function () {});
+        sync.on('error', function () {});
+        sync.on('complete', function () {
+          setTimeout(function () {
+            Object.keys(sync._events).should.have.length(0);
+            done();
+          });
+        });
+      });
+    });
   });
 });