Spinner Options

-Users can now configure default spinner behavior or provide their own.
-Provides future support for us to create other spinners that users
 can choose to deploy.
-Publishers and Subscribers with queue sizes < 1 will now have
 infinite queues. Previously you needed to disable throttling
 to get this behavior.

Author: chris-smith

src/index.js

@@ -133,7 +133,8 @@ let Rosnodejs = {
 
     // create the ros node. Return a promise that will
     // resolve when connection to master is established
-    rosNode = new RosNode(nodeName, rosMasterUri);
+    const nodeOpts = options.node || {};
+    rosNode = new RosNode(nodeName, rosMasterUri, nodeOpts);
 
 
     return this._loadOnTheFlyMessages(options)

src/lib/RosNode.js

@@ -28,7 +28,7 @@ const PublisherImpl = require('./impl/PublisherImpl.js');
 const SubscriberImpl = require('./impl/SubscriberImpl.js');
 let ServiceClient = require('./ServiceClient.js');
 let ServiceServer = require('./ServiceServer.js');
-const Spinner = require('../utils/GlobalSpinner.js');
+const GlobalSpinner = require('../utils/spinners/GlobalSpinner.js');
 let NetworkUtils = require('../utils/network_utils.js');
 let messageUtils = require('../utils/message_utils.js');
 let tcprosUtils = require('../utils/tcpros_utils.js');
@@ -78,7 +78,7 @@ class RosNode extends EventEmitter {
 
     this._setupExitHandler();
 
-    this._spinner = new Spinner();
+    this._setupSpinner(options.spinner);
   }
 
   getLogger() {
@@ -582,6 +582,30 @@ class RosNode extends EventEmitter {
     this._log.error('Not implemented');
   }
 
+  /**
+   * Initializes the spinner for this node.
+   * @param [spinnerOpts] {object} either an instance of a spinner to use or the parameters to configure one
+   * @param [spinnerOpts.type] {string} type of spinner to create
+   */
+  _setupSpinner(spinnerOpts) {
+    if (spinnerOpts) {
+      const { type } = spinnerOpts;
+      switch (type) {
+        case 'Global':
+          this._spinner = new GlobalSpinner(spinnerOpts);
+          break;
+        default:
+          // if the above didn't work, assume they created their own spinner.
+          // just use it.
+          this._spinner = spinnerOpts;
+          break;
+      }
+    }
+    else {
+      this._spinner = new GlobalSpinner();
+    }
+  }
+
   // HAVEN'T TESTED YET
   _setupExitHandler() {
     // we need to catch that this process is about to exit so we can unregister all our

src/utils/spinners/ClientQueue.js

@@ -0,0 +1,53 @@
+const EventEmitter = require('events')
+
+/**
+ * @class ClientQueue
+ * Queue of messages to handle for an individual client (subscriber or publisher)
+ */
+class ClientQueue extends EventEmitter {
+  constructor(client, queueSize, throttleMs)  {
+    super();
+
+    if (queueSize < 1) {
+      queueSize = Number.POSITIVE_INFINITY;
+    }
+
+    this._client = client;
+
+    this._queue = [];
+    this._queueSize = queueSize;
+
+    this.throttleMs = throttleMs;
+    this._handleTime = null;
+  }
+
+  destroy() {
+    this._queue = [];
+    this._client = null;
+    this._handleTime = null;
+  }
+
+  push(item) {
+    this._queue.push(item);
+    if (this.length > this._queueSize) {
+      this._queue.shift();
+    }
+  }
+
+  get length() {
+    return this._queue.length;
+  }
+
+  handleClientMessages(time) {
+    if (this._handleTime === null || time - this._handleTime >= this.throttleMs) {
+      this._handleTime = time;
+      this._client._handleMsgQueue(this._queue);
+      this._queue = [];
+      return true;
+    }
+    // else
+    return false;
+  }
+}
+
+module.exports = ClientQueue;

src/utils/GlobalSpinner.js src/utils/spinners/GlobalSpinner.jssrc/utils/GlobalSpinner.js renamed to src/utils/spinners/GlobalSpinner.js

@@ -2,53 +2,14 @@
 
 const DEFAULT_SPIN_RATE_HZ = 200;
 const events = require('events');
-const LoggingManager = require('../lib/Logging.js');
+const LoggingManager = require('../../lib/Logging.js');
 const log = LoggingManager.getLogger('ros.spinner');
 
+const ClientQueue = require('./ClientQueue.js');
+
 const PING_OP = 'ping';
 const DELETE_OP = 'delete';
-
-/**
- * @class ClientQueue
- * Queue of messages to handle for an individual client (subscriber or publisher)
- */
-class ClientQueue {
-  constructor(client, queueSize, throttleMs) {
-    if (queueSize < 1) {
-      throw new Error(`Unable to create client message queue with size ${queueSize} - minimum is 1`);
-    }
-
-    this._client = client;
-
-    this._queue = [];
-    this._queueSize = queueSize;
-
-    this.throttleMs = throttleMs;
-    this._handleTime = null;
-  }
-
-  push(item) {
-    this._queue.push(item);
-    if (this.length > this._queueSize) {
-      this._queue.shift();
-    }
-  }
-
-  get length() {
-    return this._queue.length;
-  }
-
-  handleClientMessages(time) {
-    if (this._handleTime === null || time - this._handleTime >= this.throttleMs) {
-      this._handleTime = time;
-      this._client._handleMsgQueue(this._queue);
-      this._queue = [];
-      return true;
-    }
-    // else
-    return false;
-  }
-}
+const ADD_OP = 'add';
 
 /**
  * @class GlobalSpinner
@@ -66,15 +27,16 @@ class ClientQueue {
  * ping and disconnect operations are replayed in order.
  */
 class GlobalSpinner extends events {
-  constructor(spinRate=DEFAULT_SPIN_RATE_HZ, emit=false) {
+  constructor({spinRate=null, emit=false} = {}) {
     super();
 
-    if (typeof spinRate !== 'number') {
-      spinRate = DEFAULT_SPIN_RATE_HZ;
+    if (typeof spinRate === 'number') {
+      this._spinTime = 1 / spinRate;
+    }
+    else {
+      this._spinTime = 0;
     }
 
-    this._spinTime = 1 / spinRate;
-    this._expectedSpinExpire = null;
     this._spinTimer = null;
 
     this._clientCallQueue = [];
@@ -92,8 +54,21 @@ class GlobalSpinner extends events {
     this._emit = emit;
   }
 
+  clear()  {
+    clearTimeout(this._spinTimer);
+    this._queueLocked = false;
+    this._clientQueueMap.forEach((clientQueue) => {
+      clientQueue.destroy();
+    });
+    this._clientQueueMap.clear();
+    this._clientCallQueue = [];
+  }
+
   addClient(client, clientId, queueSize, throttleMs) {
-    if (queueSize > 0) {
+    if (this._queueLocked) {
+      this._lockedOpCache.push({op: ADD_OP, client, clientId, queueSize, throttleMs});
+    }
+    else if (queueSize > 0) {
       this._clientQueueMap.set(clientId, new ClientQueue(client, queueSize, throttleMs));
     }
   }
@@ -147,28 +122,20 @@ class GlobalSpinner extends events {
   _handleLockedOpCache() {
     const len = this._lockedOpCache.length;
     for (let i = 0; i < len; ++i) {
-      const {op, clientId, msg} = this._lockedOpCache[i];
+      const {op, clientId, msg, client, queueSize, throttleMs} = this._lockedOpCache[i];
       if (op === PING_OP) {
         this.ping(clientId, msg);
       }
       else if (op === DELETE_OP) {
         this.disconnect(clientId);
       }
+      else if (op === ADD_OP) {
+        this.addClient(client, clientId, queueSize, throttleMs);
+      }
     }
     this._lockedOpCache = [];
   }
 
-  _getClientsWithQueuedMessages() {
-    const clients = {};
-    this._clientQueueMap.forEach((value, clientId) => {
-      const queueSize = value.length;
-      clients[clientId] = queueSize;
-      if (queueSize > 0 && this._clientCallQueue.indexOf(clientId) === -1) {
-        throw new Error(`Client ${clientId} has ${value.length} queued messages but is not in call list!`);
-      }
-    });
-  }
-
   _setTimer() {
     if (this._spinTimer === null) {
       if (this._emit) {
@@ -180,7 +147,6 @@ class GlobalSpinner extends events {
       else {
         this._spinTimer = setTimeout(this._handleQueue.bind(this), this._spinTime);
       }
-      this._expectedSpinExpire = Date.now() + this._spinTime;
     }
   }
 
@@ -223,4 +189,4 @@ class GlobalSpinner extends events {
   }
 }
 
-module.exports = GlobalSpinner;
+module.exports = GlobalSpinner;

test/SpinnerTest.js

@@ -2,7 +2,7 @@
 
 const chai = require('chai');
 const expect = chai.expect;
-const Spinner = require('../src/utils/GlobalSpinner.js');
+const GlobalSpinner = require('../src/utils/spinners/GlobalSpinner.js');
 
 let handleList = [];
 let spinner;
@@ -19,10 +19,10 @@ class DummyClient {
   }
 }
 
-describe('Spinner', () => {
+describe('GlobalSpinner', () => {
 
   beforeEach(() => {
-    spinner = new Spinner(null, true);
+    spinner = new GlobalSpinner({emit: true});
     handleList = [];
   });
 
@@ -122,8 +122,47 @@ describe('Spinner', () => {
           expect(spinner._lockedOpCache.length).to.equal(0);
           expect(spinner._clientCallQueue.length).to.equal(0);
           expect(spinner._clientQueueMap.has(client.id)).to.be.false;
+          handleList = [];
 
-          done();
+          spinner.addClient(client, client.id, 3, 0);
+          spinner.ping(client.id, 'junk');
+
+          expect(spinner._lockedOpCache.length).to.equal(0);
+
+          // lock the queue so the next disconnect is cached
+          spinner._queueLocked = true;
+
+          spinner.disconnect(client.id);
+          spinner.addClient(client, client.id, 3, 0);
+          spinner.ping(client.id, 'junk');
+          spinner.ping(client.id, 'junk');
+
+          expect(spinner._lockedOpCache.length).to.equal(4);
+
+          spinner.once('tick', () => {
+            // things that got in before we locked the spinner
+            expect(handleList.length).to.equal(1);
+            expect(handleList[0].queue.length).to.equal(1);
+            expect(spinner._lockedOpCache.length).to.equal(0);
+            expect(spinner._clientCallQueue.length).to.equal(1);
+            expect(spinner._clientQueueMap.has(client.id)).to.be.true;
+            handleList = [];
+
+            // things that got in after we locked the spinner
+            spinner.once('tick', () => {
+              expect(handleList.length).to.equal(1);
+              expect(handleList[0].queue.length).to.equal(2);
+              expect(spinner._lockedOpCache.length).to.equal(0);
+              expect(spinner._clientCallQueue.length).to.equal(0);
+              expect(spinner._clientQueueMap.has(client.id)).to.be.true;
+
+              spinner.disconnect(client.id);
+
+              expect(spinner._clientQueueMap.has(client.id)).to.be.false;
+
+              done();
+            });
+          });
         });
       });
     });
@@ -143,7 +182,7 @@ describe('Spinner', () => {
       spinner.on('tick', () => {
         if (spinner._clientCallQueue.length === 0) {
           const lastTick = Date.now();
-          const tDiff = lastTick - firstTick;
+          const tDiff = lastTick - firstTick + 1;
           expect(tDiff).to.be.at.least(throttleMs);
 
           done();

test/xmlrpcTest.js

@@ -51,6 +51,8 @@ describe('Protocol Test', () => {
       nh._node._subscribers = {};
       nh._node._publishers = {};
 
+      nh._node._spinner.clear();
+
       // remove any master api handlers we set up
       masterStub.removeAllListeners();
     });
@@ -307,6 +309,8 @@ describe('Protocol Test', () => {
       nh._node._subscribers = {};
       nh._node._publishers = {};
 
+      nh._node._spinner.clear();
+
       // remove any master api handlers we set up
       masterStub.removeAllListeners();
     });
@@ -720,6 +724,8 @@ describe('Protocol Test', () => {
       nh._node._subscribers = {};
       nh._node._publishers = {};
 
+      nh._node._spinner.clear();
+
       // remove any master api handlers we set up
       masterStub.removeAllListeners();
     });