fix: Merge pull request #25 from 6RiverSystems/feat/latch-subs

fix: latched subs

Author: dnikitop

src/lib/RosNode.js

@@ -111,15 +111,22 @@ class RosNode extends EventEmitter {
   subscribe(options, callback) {
     let topic = options.topic;
     let subImpl = this._subscribers[topic];
+    let firstSubscriber = false;
     if (!subImpl) {
       subImpl = new SubscriberImpl(options, this);
       this._subscribers[topic] = subImpl;
+      firstSubscriber = true;
     }
 
     const sub = new Subscriber(subImpl);
     if (callback && typeof callback === 'function') {
       sub.on('message', callback);
     }
+    // Duplicate subscribers wont get a latched 'message' emit from the existing subscriber impl.
+    // This will forcibly send an emit if the topic is latched and has already seen an incoming message
+    if (!firstSubscriber && subImpl.getLatching() && subImpl.getLastMessage()) {
+        sub.emit('message', subImpl.getLastMessage())
+    }
 
     return sub;
   }

src/lib/impl/SubscriberImpl.js

@@ -85,6 +85,11 @@ class SubscriberImpl extends EventEmitter {
     this._pendingPubClients = {};
 
     this._state = REGISTERING;
+
+    this._latching = undefined;
+
+    this._lastMessage = undefined;
+  
     this._register();
   }
 
@@ -129,6 +134,23 @@ class SubscriberImpl extends EventEmitter {
     return this._nodeHandle;
   }
 
+  /**
+   * Check if this subscriber is connected to a latched topic
+   * @returns {boolean}
+   */
+  getLatching() {
+    return this._latching;
+  }
+
+  /**
+   * Return the last message received
+   * @returns {any}
+   */
+  getLastMessage() {
+    return this._lastMessage;
+  }
+
+
   /**
    * Clears and closes all client connections for this subscriber.
    */
@@ -389,6 +411,10 @@ class SubscriberImpl extends EventEmitter {
     // remove client from pending map now that it's validated
     delete this._pendingPubClients[client.nodeUri];
 
+    if (header.latching) {
+      this._latching = true;
+    }
+
     // pipe all future messages to _handleMessage
     client.$deserializer.on('message', this._handleMessage.bind(this));
 
@@ -416,6 +442,9 @@ class SubscriberImpl extends EventEmitter {
   _handleMsgQueue(msgQueue) {
     try {
       msgQueue.forEach((msg) => {
+        if (this.getLatching()) {
+          this._lastMessage = this._messageHandler.deserialize(msg);
+        }
         this.emit('message', this._messageHandler.deserialize(msg));
       });
     }

test/xmlrpcTest.js

@@ -739,6 +739,30 @@ describe('Protocol Test', () => {
         });
       });
     });
+
+    it('2 Subscribers on Same Latched Topic subscribing at different times ', function(done) {
+      this.slow(1000);
+      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) => {
+          expect(sub1._impl.listenerCount('connection')).to.equal(2);
+          msg2 = msg.data;
+          expect(msg1).to.equal(1);
+          expect(msg1).to.equal(msg2);
+          done()
+        });
+      });
+
+      expect(sub1._impl.listenerCount('connection')).to.equal(1);
+
+      const pub = nh.advertise(topic, msgType, {latching: true});
+
+      pub.publish({data: 1});
+    });
   });
 
   describe('Service', () => {