Subscriber shutdown bug

Fixes an issue where publisher clients in the process
of being validated will not be destroyed when the
subscriber is shutdown.

Author: chris-smith

src/lib/impl/PublisherImpl.js

@@ -154,7 +154,6 @@ class PublisherImpl extends EventEmitter {
     Object.keys(this._subClients).forEach((clientId) => {
       const client = this._subClients[clientId];
       client.end();
-      client.destroy();
     });
 
     // disconnect from the spinner in case we have any pending callbacks
@@ -274,19 +273,21 @@ class PublisherImpl extends EventEmitter {
     }
 
     subscriber.on('close', () => {
-      this._log.info('Publisher %s client %s disconnected!',
-                      this.getTopic(), subscriber.name);
+      this._log.info('Publisher client socket %s on topic %s disconnected',
+                     subscriber.name, this.getTopic());
       subscriber.removeAllListeners();
       delete this._subClients[subscriber.name];
       this.emit('disconnect');
     });
 
     subscriber.on('end', () => {
-      this._log.info('Sub %s sent END', subscriber.name);
+      this._log.info('Publisher client socket %s on topic %s ended the connection',
+                     subscriber.name, this.getTopic());
     });
 
-    subscriber.on('error', () => {
-      this._log.warn('Sub %s had error', subscriber.name);
+    subscriber.on('error', (err) => {
+      this._log.warn('Publisher client socket %s on topic %s had error: %s',
+                     subscriber.name, this.getTopic(), err);
     });
 
     // if we've cached a message from latching, send it now

src/lib/impl/SubscriberImpl.js

@@ -78,6 +78,8 @@ class SubscriberImpl extends EventEmitter {
 
     this._pubClients = {};
 
+    this._pendingPubClients = {};
+
     this._state = REGISTERING;
     this._register();
   }
@@ -131,10 +133,12 @@ class SubscriberImpl extends EventEmitter {
     this._log.debug('Shutting down subscriber %s', this.getTopic());
 
     Object.keys(this._pubClients).forEach(this._disconnectClient.bind(this));
+    Object.keys(this._pendingPubClients).forEach(this._disconnectClient.bind(this));
 
     // disconnect from the spinner in case we have any pending callbacks
     this._nodeHandle.getSpinner().disconnect(this._getSpinnerId());
     this._pubClients = {};
+    this._pendingPubClients = {};
   }
 
   /**
@@ -210,11 +214,16 @@ class SubscriberImpl extends EventEmitter {
    * @param clientId {string}
    */
   _disconnectClient(clientId) {
-    const client = this._pubClients[clientId];
+    let client = this._pubClients[clientId];
+
+    const hasValidatedClient = !!client;
+    if (!hasValidatedClient) {
+      client = this._pendingPubClients[clientId];
+    }
+
     if (client) {
-      this._log.debug('Disconnecting client %s', clientId);
+      this._log.debug('Subscriber %s disconnecting client %s', this.getTopic(), clientId);
       client.end();
-      client.destroy();
 
       client.removeAllListeners();
       client.$deserializer.removeAllListeners();
@@ -223,11 +232,13 @@ class SubscriberImpl extends EventEmitter {
       client.unpipe(client.$deserializer);
 
       delete client.$deserializer;
-      delete client.$boundMessageHandler;
 
       delete this._pubClients[clientId];
+      delete this._pendingPubClients[clientId];
 
-      this.emit('disconnect');
+      if (hasValidatedClient) {
+        this.emit('disconnect');
+      }
     }
   }
 
@@ -285,11 +296,20 @@ class SubscriberImpl extends EventEmitter {
     client.nodeUri = nodeUri;
 
     client.on('end', () => {
-      this._log.info('Pub %s sent END', client.name, this.getTopic());
+      this._log.info('Subscriber client socket %s on topic %s ended the connection',
+                     client.name, this.getTopic());
     });
 
-    client.on('error', () => {
-      this._log.warn('Pub %s error on topic %s', client.name, this.getTopic());
+    client.on('error', (err) => {
+      this._log.warn('Subscriber client socket %s on topic %s had error: %s',
+                     client.name, this.getTopic(), err);
+    });
+
+    // hook into close event to clean things up
+    client.on('close', () => {
+      this._log.info('Subscriber client socket %s on topic %s disconnected',
+                     client.name, this.getTopic());
+      this._disconnectClient(client.nodeUri);
     });
 
     // open the socket at the provided address, port
@@ -308,6 +328,11 @@ class SubscriberImpl extends EventEmitter {
     client.$deserializer = deserializer;
     client.pipe(deserializer);
 
+    // cache client in "pending" map.
+    // It's not validated yet so we don't want it to show up as a client.
+    // Need to keep track of it in case we're shutdown before it can be validated.
+    this._pendingPubClients[client.nodeUri] = client;
+
     // create a one-time handler for the connection header
     // if the connection is validated, we'll listen for more events
     deserializer.once('message', this._handleConnectionHeader.bind(this, client));
@@ -329,6 +354,11 @@ class SubscriberImpl extends EventEmitter {
    * @param msg {string} message received from the publisher
    */
   _handleConnectionHeader(client, msg) {
+    if (this.isShutdown()) {
+      this._disconnectClient(client.nodeUri);
+      return;
+    }
+
     let header = TcprosUtils.parseTcpRosHeader(msg);
     // check if the publisher had a problem with our connection header
     if (header.error) {
@@ -349,18 +379,13 @@ class SubscriberImpl extends EventEmitter {
 
     // cache client now that we've verified the connection header
     this._pubClients[client.nodeUri] = client;
+    // remove client from pending map now that it's validated
+    delete this._pendingPubClients[client.nodeUri];
 
     // pipe all future messages to _handleMessage
     client.$deserializer.on('message', this._handleMessage.bind(this));
 
     this.emit('connection', header, client.name);
-
-    // hook into close event to clean things up
-    client.on('close', () => {
-      this._log.warn('Pub %s closed on topic %s', client.name, this.getTopic());
-      this._log.warn('Subscriber ' + this.getTopic() + ' client ' + client.name + ' disconnected!');
-      this._disconnectClient(client.nodeUri);
-    });
   }
 
   /**

test/xmlrpcTest.js

@@ -566,7 +566,7 @@ describe('Protocol Test', () => {
           SOCKET_END_CACHED.call(this);
         };
 
-        sub._handleTopicRequestResponse([1, 'ok', ['TCPROS', 'junk_address', 1234]], 'http://junk_address:1234');
+        subImpl._handleTopicRequestResponse([1, 'ok', ['TCPROS', 'junk_address', 1234]], 'http://junk_address:1234');
         sub.shutdown();
       });
     });
@@ -594,7 +594,7 @@ describe('Protocol Test', () => {
       });
       let pub = nh.advertise(topic, msgType);
 
-      pub.on('connected', () => {
+      pub.on('connection', () => {
         pub.publish({data: 1});
         pub.shutdown();
       });
@@ -603,6 +603,37 @@ describe('Protocol Test', () => {
       setTimeout(done, 500);
     });
 
+    it('Shutdown Subscriber With Pending Publisher Client', function(done) {
+      this.slow(1600);
+      const nh = rosnodejs.nh;
+      const sub = nh.subscribe(topic, msgType, () => {
+        throwNext('Subscriber should never have gotten messages!');
+      });
+      let pub = nh.advertise(topic, msgType);
+
+      // when publisher emits 'connection', it has validated
+      // the subscriber's connection header and sent a response back
+      // the subscriber will not have validated the publisher's connection
+      // header though so it should have a pending publisher entry.
+      pub.on('connection', () => {
+        const impl = sub._impl;
+
+        expect(Object.keys(impl._pendingPubClients)).to.have.lengthOf(1);
+        expect(impl._pubClients).to.be.empty;
+
+        sub.shutdown();
+
+        expect(impl._pendingPubClients).to.be.empty;
+        expect(impl._pubClients).to.be.empty;
+
+        setTimeout(() => {
+          expect(impl._pendingPubClients).to.be.empty;
+          expect(impl._pubClients).to.be.empty;
+          done();
+        }, 500);
+      });
+    });
+
     it('2 Publishers on Same Topic', function(done) {
       this.slow(2000);
       const nh = rosnodejs.nh;