Fix stale listener bug

-Fixes an issue with stale listeners
 on pub, sub impls.
-Adds tests to check for stale listeners.

Author: chris-smith

package.json

@@ -34,6 +34,7 @@
     "bunyan": "1.8.1",
     "md5": "2.1.0",
     "moment": "2.12.0",
+    "ultron": "1.1.0",
     "walker": "1.0.7",
     "xmlrpc": "chfritz/node-xmlrpc"
   }

src/lib/Publisher.js

@@ -18,6 +18,7 @@
 "use strict";
 
 const EventEmitter = require('events');
+const Ultron = require('ultron');
 const {rebroadcast} = require('../utils/event_utils.js');
 
 /**
@@ -31,14 +32,15 @@ class Publisher extends EventEmitter {
 
     ++impl.count;
     this._impl = impl;
+    this._ultron = new Ultron(impl);
 
     this._topic = impl.getTopic();
     this._type = impl.getType();
 
-    rebroadcast('registered', this._impl, this);
-    rebroadcast('connection', this._impl, this);
-    rebroadcast('disconnect', this._impl, this);
-    rebroadcast('error', this._impl, this);
+    rebroadcast('registered', this._ultron, this);
+    rebroadcast('connection', this._ultron, this);
+    rebroadcast('disconnect', this._ultron, this);
+    rebroadcast('error', this._ultron, this);
   }
 
   /**
@@ -92,6 +94,8 @@ class Publisher extends EventEmitter {
     if (this._impl) {
       const impl = this._impl
       this._impl = null;
+      this._ultron.destroy();
+      this._ultron = null;
 
       --impl.count;
       if (impl.count <= 0) {

src/lib/Subscriber.js

@@ -18,6 +18,7 @@
 "use strict";
 
 const EventEmitter = require('events');
+const Ultron = require('ultron');
 const {rebroadcast} = require('../utils/event_utils.js');
 
 //-----------------------------------------------------------------------
@@ -33,15 +34,16 @@ class Subscriber extends EventEmitter {
 
     ++impl.count;
     this._impl = impl;
+    this._ultron = new Ultron(impl);
 
     this._topic = impl.getTopic();
     this._type = impl.getType();
 
-    rebroadcast('registered', this._impl, this);
-    rebroadcast('connection', this._impl, this);
-    rebroadcast('disconnect', this._impl, this);
-    rebroadcast('error', this._impl, this);
-    rebroadcast('message', this._impl, this);
+    rebroadcast('registered', this._ultron, this);
+    rebroadcast('connection', this._ultron, this);
+    rebroadcast('disconnect', this._ultron, this);
+    rebroadcast('error', this._ultron, this);
+    rebroadcast('message', this._ultron, this);
   }
 
   /**
@@ -82,6 +84,8 @@ class Subscriber extends EventEmitter {
     if (this._impl) {
       const impl = this._impl
       this._impl = null;
+      this._ultron.destroy();
+      this._ultron = null;
 
       --impl.count;
       if (impl.count <= 0) {

test/xmlrpcTest.js

@@ -579,6 +579,8 @@ describe('Protocol Test', () => {
       const pub2 = nh.advertise(topic, msgType, {latching: true});
 
       expect(pub1).to.not.equal(pub2);
+      expect(pub1._impl.listenerCount('connection')).to.equal(2);
+      expect(pub2._impl.listenerCount('connection')).to.equal(2);
 
       pub1.publish({data: 1});
 
@@ -592,6 +594,9 @@ describe('Protocol Test', () => {
 
           pub1.shutdown()
           .then(() => {
+            expect(pub1._impl).to.equal(null);
+            expect(pub2._impl.listenerCount('connection')).to.equal(1);
+
             expect(sub.getNumPublishers()).to.equal(1);
 
             pub2.publish({data: 3});
@@ -602,6 +607,7 @@ describe('Protocol Test', () => {
               pub2.shutdown()
               .then(() =>  {
                 expect(sub.getNumPublishers()).to.equal(0);
+                expect(pub2._impl).to.equal(null);
                 done();
               });
             });
@@ -624,7 +630,9 @@ describe('Protocol Test', () => {
         msg2 = msg.data;
       });
 
-      expect(sub1).to.not.equal(msg2);
+      expect(sub1).to.not.equal(sub2);
+      expect(sub1._impl.listenerCount('connection')).to.equal(2);
+      expect(sub2._impl.listenerCount('connection')).to.equal(2);
 
       const pub = nh.advertise(topic, msgType, {latching: true});
 
@@ -643,6 +651,8 @@ describe('Protocol Test', () => {
 
           sub1.shutdown()
           .then(() => {
+            expect(sub1._impl).to.equal(null);
+            expect(sub2._impl.listenerCount('connection')).to.equal(1);
             pub.publish({data: 30});
 
             sub2.once('message', () => {
@@ -652,6 +662,7 @@ describe('Protocol Test', () => {
 
               sub2.shutdown()
               .then(() => {
+                expect(sub2._impl).to.equal(null);
                 expect(pub.getNumSubscribers()).to.equal(0);
                 done();
               });