Add support for tcpNoDelay option on subscribers. Clean up (#107)

Author: chris-smith

src/lib/impl/PublisherImpl.js

@@ -43,8 +43,7 @@ class PublisherImpl extends EventEmitter {
 
     this._tcpNoDelay =  !!options.tcpNoDelay;
 
-
-    if (options.queueSize) {
+    if (options.hasOwnProperty('queueSize')) {
       this._queueSize = options.queueSize;
     }
     else {
@@ -165,7 +164,7 @@ class PublisherImpl extends EventEmitter {
   }
 
   /**
-   * Check if this subscriber has been shutdown
+   * Check if this publisher has been shutdown
    * @returns {boolean}
    */
   isShutdown() {
@@ -242,18 +241,18 @@ class PublisherImpl extends EventEmitter {
   /**
    * Handles a new connection from a subscriber to this publisher's node.
    * Validates the connection header and sends a response header
-   * @param subscriber {Socket} client that sent the header
+   * @param socket {Socket} client that sent the header
    * @param header {Object} deserialized connection header.
    */
-  handleSubscriberConnection(subscriber, header) {
+  handleSubscriberConnection(socket, header) {
     let error = TcprosUtils.validateSubHeader(
       header, this.getTopic(), this.getType(),
       this._messageHandler.md5sum());
 
     if (error !== null) {
       this._log.error('Unable to validate subscriber connection header '
                       + JSON.stringify(header));
-      subscriber.end(Serialize(error));
+      socket.end(Serialize(error));
       return;
     }
     // else
@@ -267,42 +266,42 @@ class PublisherImpl extends EventEmitter {
         this.getType(),
         this.getLatching(),
         this._messageHandler.messageDefinition());
-    subscriber.write(respHeader);
+    socket.write(respHeader);
 
     // if this publisher had the tcpNoDelay option set
     // disable the nagle algorithm
-    if  (this._tcpNoDelay || header.tcp_nodelay === 1) {
-      subscriber.setNoDelay(true);
+    if  (this._tcpNoDelay || header.tcp_nodelay === '1') {
+      socket.setNoDelay(true);
     }
 
-    subscriber.on('close', () => {
+    socket.on('close', () => {
       this._log.info('Publisher client socket %s on topic %s disconnected',
-                     subscriber.name, this.getTopic());
-      subscriber.removeAllListeners();
-      delete this._subClients[subscriber.name];
+                     socket.name, this.getTopic());
+      socket.removeAllListeners();
+      delete this._subClients[socket.name];
       this.emit('disconnect');
     });
 
-    subscriber.on('end', () => {
+    socket.on('end', () => {
       this._log.info('Publisher client socket %s on topic %s ended the connection',
-                     subscriber.name, this.getTopic());
+                     socket.name, this.getTopic());
     });
 
-    subscriber.on('error', (err) => {
+    socket.on('error', (err) => {
       this._log.warn('Publisher client socket %s on topic %s had error: %s',
-                     subscriber.name, this.getTopic(), err);
+                     socket.name, this.getTopic(), err);
     });
 
     // if we've cached a message from latching, send it now
     if (this._lastSentMsg !== null) {
       this._log.debug('Sending latched msg to new subscriber');
-      subscriber.write(this._lastSentMsg);
+      socket.write(this._lastSentMsg);
     }
 
     // handshake was good - we'll start publishing to it
-    this._subClients[subscriber.name] = subscriber;
+    this._subClients[socket.name] = socket;
 
-    this.emit('connection', header, subscriber.name);
+    this.emit('connection', header, socket.name);
   }
 
   /**

src/lib/impl/SubscriberImpl.js

@@ -48,7 +48,7 @@ class SubscriberImpl extends EventEmitter {
 
     this._type = options.type;
 
-    if (options.queueSize) {
+    if (options.hasOwnProperty('queueSize')) {
       this._queueSize = options.queueSize;
     }
     else {
@@ -67,6 +67,10 @@ class SubscriberImpl extends EventEmitter {
       this._throttleMs = 0;
     }
 
+    // tcpNoDelay will be set as a field in the connection header sent to the
+    // relevant publisher - the publisher should then set tcpNoDelay on the socket
+    this._tcpNoDelay =  !!options.tcpNoDelay;
+
     this._msgHandleTime = null;
 
     this._nodeHandle = nodeHandle;
@@ -294,51 +298,51 @@ class SubscriberImpl extends EventEmitter {
     let port = info[2];
     let address = info[1];
 
-    let client = new Socket();
-    client.name = address + ':' + port;
-    client.nodeUri = nodeUri;
+    let socket = new Socket();
+    socket.name = address + ':' + port;
+    socket.nodeUri = nodeUri;
 
-    client.on('end', () => {
+    socket.on('end', () => {
       this._log.info('Subscriber client socket %s on topic %s ended the connection',
-                     client.name, this.getTopic());
+                     socket.name, this.getTopic());
     });
 
-    client.on('error', (err) => {
+    socket.on('error', (err) => {
       this._log.warn('Subscriber client socket %s on topic %s had error: %s',
-                     client.name, this.getTopic(), err);
+                     socket.name, this.getTopic(), err);
     });
 
     // hook into close event to clean things up
-    client.on('close', () => {
+    socket.on('close', () => {
       this._log.info('Subscriber client socket %s on topic %s disconnected',
-                     client.name, this.getTopic());
-      this._disconnectClient(client.nodeUri);
+                     socket.name, this.getTopic());
+      this._disconnectClient(socket.nodeUri);
     });
 
     // open the socket at the provided address, port
-    client.connect(port, address, () => {
+    socket.connect(port, address, () => {
       if (this.isShutdown()) {
-        client.end();
+        socket.end();
         return;
       }
 
       this._log.debug('Subscriber on ' + this.getTopic() + ' connected to publisher at ' + address + ':' + port);
-      client.write(this._createTcprosHandshake());
+      socket.write(this._createTcprosHandshake());
     });
 
     // create a DeserializeStream to chunk out messages
     let deserializer = new DeserializeStream();
-    client.$deserializer = deserializer;
-    client.pipe(deserializer);
+    socket.$deserializer = deserializer;
+    socket.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;
+    this._pendingPubClients[socket.nodeUri] = socket;
 
     // 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));
+    deserializer.once('message', this._handleConnectionHeader.bind(this, socket));
   }
 
   /**
@@ -347,18 +351,18 @@ class SubscriberImpl extends EventEmitter {
    */
   _createTcprosHandshake() {
     return TcprosUtils.createSubHeader(this._nodeHandle.getNodeName(), this._messageHandler.md5sum(),
-                                       this.getTopic(), this.getType(), this._messageHandler.messageDefinition());
+                                       this.getTopic(), this.getType(), this._messageHandler.messageDefinition(), this._tcpNoDelay);
   }
 
   /**
    * Handles the connection header from a publisher. If connection is validated,
    * we'll start handling messages from the client.
-   * @param client {Socket} publisher client who sent the connection header
+   * @param socket {Socket} publisher client who sent the connection header
    * @param msg {string} message received from the publisher
    */
-  _handleConnectionHeader(client, msg) {
+  _handleConnectionHeader(socket, msg) {
     if (this.isShutdown()) {
-      this._disconnectClient(client.nodeUri);
+      this._disconnectClient(socket.nodeUri);
       return;
     }
 
@@ -373,22 +377,21 @@ class SubscriberImpl extends EventEmitter {
     const error = TcprosUtils.validatePubHeader(header, this.getType(), this._messageHandler.md5sum());
     if (error) {
       this._log.error(`Unable to validate subscriber ${this.getTopic()} connection header ${JSON.stringify(header)}`);
-      TcprosUtils.parsePubHeader(msg);
-      client.end(Serialize(error));
+      socket.end(Serialize(error));
       return;
     }
     // connection header was valid - we're good to go!
     this._log.debug('Subscriber ' + this.getTopic() + ' got connection header ' + JSON.stringify(header));
 
     // cache client now that we've verified the connection header
-    this._pubClients[client.nodeUri] = client;
+    this._pubClients[socket.nodeUri] = socket;
     // remove client from pending map now that it's validated
-    delete this._pendingPubClients[client.nodeUri];
+    delete this._pendingPubClients[socket.nodeUri];
 
     // pipe all future messages to _handleMessage
-    client.$deserializer.on('message', this._handleMessage.bind(this));
+    socket.$deserializer.on('message', this._handleMessage.bind(this));
 
-    this.emit('connection', header, client.name);
+    this.emit('connection', header, socket.name);
   }
 
   /**

src/utils/tcpros_utils.js

@@ -28,10 +28,11 @@ const md5Prefix = 'md5sum=';
 const topicPrefix = 'topic=';
 const servicePrefix = 'service=';
 const typePrefix = 'type=';
-const latchingPrefix = 'latching=';
-const persistentPrefix = 'persistent=';
 const errorPrefix = 'error=';
 const messageDefinitionPrefix = 'message_definition=';
+const latchingField = 'latching=1';
+const persistentField = 'persistent=1';
+const tcpNoDelayField = 'tcp_nodelay=1';
 
 //-----------------------------------------------------------------------
 
@@ -66,14 +67,19 @@ function deserializeStringFields(buffer) {
  */
 let TcprosUtils = {
 
-  createSubHeader(callerId, md5sum, topic, type, messageDefinition) {
+  createSubHeader(callerId, md5sum, topic, type, messageDefinition, tcpNoDelay) {
     const fields = [
       callerIdPrefix + callerId,
       md5Prefix + md5sum,
       topicPrefix + topic,
       typePrefix + type,
       messageDefinitionPrefix + messageDefinition
     ];
+
+    if (tcpNoDelay) {
+      fields.push(tcpNoDelayField);
+    }
+
     return serializeStringFields(fields);
   },
 
@@ -92,9 +98,13 @@ let TcprosUtils = {
       callerIdPrefix + callerId,
       md5Prefix + md5sum,
       typePrefix + type,
-      latchingPrefix + latching,
       messageDefinitionPrefix + messageDefinition
     ];
+
+    if (latching) {
+      fields.push(latchingField);
+    }
+
     return serializeStringFields(fields);
   },
 
@@ -106,8 +116,9 @@ let TcprosUtils = {
     ];
 
     if (persistent) {
-      fields.push(persistentPrefix + '1');
+      fields.push(persistentField);
     }
+
     return serializeStringFields(fields);
   },
 
@@ -139,87 +150,6 @@ let TcprosUtils = {
     return info;
   },
 
-  parseSubHeader(header) {
-    let info = {};
-    const fields = deserializeStringFields(header);
-    fields.forEach((field) => {
-
-      if (field.startsWith(md5Prefix)) {
-        info.md5sum = field.substr(md5Prefix.length);
-      }
-      else if (field.startsWith(topicPrefix)) {
-        info.topic = field.substr(topicPrefix.length);
-      }
-      else if (field.startsWith(callerIdPrefix)) {
-        info.callerId = field.substr(callerIdPrefix.length);
-      }
-      else if (field.startsWith(typePrefix)) {
-        info.type = field.substr(typePrefix.length);
-      }
-    });
-    return info;
-  },
-
-  parsePubHeader(header) {
-    let info = {};
-    const fields = deserializeStringFields(header);
-    fields.forEach((field) => {
-
-      if (field.startsWith(md5Prefix)) {
-        info.md5sum = field.substr(md5Prefix.length);
-      }
-      else if (field.startsWith(latchingPrefix)) {
-        info.latching = field.substr(latchingPrefix.length);
-      }
-      else if (field.startsWith(callerIdPrefix)) {
-        info.callerId = field.substr(callerIdPrefix.length);
-      }
-      else if (field.startsWith(typePrefix)) {
-        info.type = field.substr(typePrefix.length);
-      }
-    });
-    return info;
-  },
-
-  parseServiceClientHeader(header) {
-    let info = {};
-    const fields = deserializeStringFields(header);
-    fields.forEach((field) => {
-
-      if (field.startsWith(md5Prefix)) {
-        info.md5sum = field.substr(md5Prefix.length);
-      }
-      else if (field.startsWith(servicePrefix)) {
-        info.service = field.substr(servicePrefix.length);
-      }
-      else if (field.startsWith(callerIdPrefix)) {
-        info.callerId = field.substr(callerIdPrefix.length);
-      }
-      else if (field.startsWith(typePrefix)) {
-        info.type = field.substr(typePrefix.length);
-      }
-    });
-    return info;
-  },
-
-  parseServiceServerHeader(header) {
-    let info = {};
-    const fields = deserializeStringFields(header);
-    fields.forEach((field) => {
-
-      if (field.startsWith(md5Prefix)) {
-        info.md5sum = field.substr(md5Prefix.length);
-      }
-      else if (field.startsWith(callerIdPrefix)) {
-        info.callerId = field.substr(callerIdPrefix.length);
-      }
-      else if (field.startsWith(typePrefix)) {
-        info.type = field.substr(typePrefix.length);
-      }
-    });
-    return info;
-  },
-
   validateSubHeader(header, topic, type, md5sum) {
     if (!header.hasOwnProperty('topic')) {
       return this.serializeString('Connection header missing expected field [topic]');

test/onTheFly.js

@@ -11,7 +11,7 @@ const MASTER_PORT = 11234;
 describe('OnTheFly', function () {
   let master;
 
-  before(() => {
+  before(function() {
     this.timeout(0);
 
     master = new Master('localhost', MASTER_PORT);