Shutdown logging cleanup on shutdown. Add tests for multiple pub/subs. Don't advertise the same service twice per node.

Author: chris-smith

src/lib/Logging.js

@@ -239,6 +239,10 @@ class LoggingManager {
     this._forEachLogger((logger) => logger.clearExpiredThrottledLogs(), true);
   }
 
+  stopLogCleanup() {
+    clearInterval(this._cleanLoggersInterval);
+  }
+
   _handleGetLoggers(req, resp) {
     if (this._externalLog.getLoggers !== null) {
       this._externalLog.getLoggers(req, resp);

src/lib/RosNode.js

@@ -123,10 +123,13 @@ class RosNode extends EventEmitter {
   advertiseService(options, callback) {
     let service = options.service;
     let serv = this._services[service];
-    if (!serv) {
-      serv = new ServiceServer(options, callback, this);
-      this._services[service] = serv;
+    if (serv) {
+      this._log.warn('Tried to advertise a service that is already advertised in this node [%s]', service);
+      return;
     }
+    // else
+    serv = new ServiceServer(options, callback, this);
+    this._services[service] = serv;
     return serv;
   }
 
@@ -635,6 +638,8 @@ class RosNode extends EventEmitter {
       promises.push(shutdownServers());
       promises.push(clearXmlrpcQueues());
 
+      Logging.stopLogCleanup();
+
       process.removeListener('exit', exitHandler);
       process.removeListener('SIGINT', sigIntHandler);
 

src/lib/ServiceServer.js

@@ -108,6 +108,10 @@ class ServiceServer extends EventEmitter {
   }
 
   handleClientConnection(client, header) {
+    if (this.isShutdown()) {
+      return;
+    }
+    // else
     // TODO: verify header data
     this._log.debug('Service %s handling new client connection ', this.getService());
 

src/lib/impl/PublisherImpl.js

@@ -268,7 +268,7 @@ class PublisherImpl extends EventEmitter {
 
     // if this publisher had the tcpNoDelay option set
     // disable the nagle algorithm
-    if  (this._tcpNoDelay) {
+    if  (this._tcpNoDelay || header.tcp_nodelay === 1) {
       subscriber.setNoDelay(true);
     }
 

src/utils/tcpros_utils.js

@@ -143,7 +143,7 @@ let TcprosUtils = {
     let info = {};
     const fields = deserializeStringFields(header);
     fields.forEach((field) => {
-      
+
       if (field.startsWith(md5Prefix)) {
         info.md5sum = field.substr(md5Prefix.length);
       }
@@ -164,7 +164,7 @@ let TcprosUtils = {
     let info = {};
     const fields = deserializeStringFields(header);
     fields.forEach((field) => {
-      
+
       if (field.startsWith(md5Prefix)) {
         info.md5sum = field.substr(md5Prefix.length);
       }
@@ -206,7 +206,7 @@ let TcprosUtils = {
     let info = {};
     const fields = deserializeStringFields(header);
     fields.forEach((field) => {
-      
+
       if (field.startsWith(md5Prefix)) {
         info.md5sum = field.substr(md5Prefix.length);
       }

test/xmlrpcTest.js

@@ -565,6 +565,101 @@ describe('Protocol Test', () => {
       // if we haven't received a message by now we should be good
       setTimeout(done, 500);
     });
+
+    it('2 Publishers on Same Topic', function(done) {
+      this.slow(2000);
+      const nh = rosnodejs.nh;
+
+      let msg1;
+      const sub = nh.subscribe(topic, msgType, (msg) => {
+          msg1 = msg.data;
+      });
+
+      const pub1 = nh.advertise(topic, msgType, {latching: true});
+      const pub2 = nh.advertise(topic, msgType, {latching: true});
+
+      expect(pub1).to.not.equal(pub2);
+
+      pub1.publish({data: 1});
+
+      sub.once('message', ({data}) => {
+        expect(sub.getNumPublishers()).to.equal(1);
+        expect(data).to.equal(1);
+
+        pub2.publish({data: 2});
+        sub.once('message', ({data}) => {
+          expect(data).to.equal(2);
+
+          pub1.shutdown()
+          .then(() => {
+            expect(sub.getNumPublishers()).to.equal(1);
+
+            pub2.publish({data: 3});
+
+            sub.once('message', ({data}) => {
+              expect(data).to.equal(3);
+
+              pub2.shutdown()
+              .then(() =>  {
+                expect(sub.getNumPublishers()).to.equal(0);
+                done();
+              });
+            });
+          })
+        });
+      })
+    });
+
+    it('2 Subscribers on Same Topic', function(done) {
+      this.slow(2000);
+      const nh = rosnodejs.nh;
+
+      let msg1;
+      const sub1 = nh.subscribe(topic, msgType, (msg) => {
+          msg1 = msg.data;
+      });
+
+      let msg2;
+      const sub2 = nh.subscribe(topic, msgType, (msg) => {
+        msg2 = msg.data;
+      });
+
+      expect(sub1).to.not.equal(msg2);
+
+      const pub = nh.advertise(topic, msgType, {latching: true});
+
+      pub.publish({data: 1});
+
+      sub2.once('message', () => {
+        expect(pub.getNumSubscribers()).to.equal(1);
+
+        expect(msg1).to.equal(msg2);
+        pub.publish({data: 25});
+
+        sub2.once('message', () => {
+          expect(msg1).to.equal(msg2);
+          msg1 = null;
+          msg2 = null;
+
+          sub1.shutdown()
+          .then(() => {
+            pub.publish({data: 30});
+
+            sub2.once('message', () => {
+              expect(msg1).to.equal(null);
+              expect(msg2).to.equal(30);
+              expect(pub.getNumSubscribers()).to.equal(1);
+
+              sub2.shutdown()
+              .then(() => {
+                expect(pub.getNumSubscribers()).to.equal(0);
+                done();
+              });
+            });
+          });
+        });
+      });
+    });
   });
 
   describe('Service', () => {