connection: use ping-pong instead of heartbeat

lib/connection.js

@@ -2,7 +2,6 @@
 
 const { EventEmitter } = require('events');
 const semver = require('semver');
-const timers = require('timers');
 
 const common = require('./common');
 const serde = require('./serde');
@@ -53,18 +52,15 @@ class Connection extends EventEmitter {
     this.application = null;
     this.remoteProxies = {};
 
-    this._heartbeatCallbackInstance = null;
+    this._heartbeatInterval = null;
+    this._closed = false;
 
     // Defined in constructor to be used as default callback in callMethod
     // without binding it.
     this._emitError = (error) => {
       if (error) this.emit('error', error);
     };
 
-    // Defined in constructor to be used as heartbeat message
-    // in debug mode events
-    this._heartbeatMessage = {};
-
     transport.on('message', this._processMessage.bind(this));
     transport.on('close', this._onSocketClose.bind(this));
     transport.on('error', this._onSocketError.bind(this));
@@ -241,33 +237,36 @@ class Connection extends EventEmitter {
   // Send a pong message
   //
   pong(messageId) {
+    if (this._closed) {
+      return;
+    }
     const message = { pong: [messageId] };
     this._send(this._prepareMessage(message), message);
   }
 
-  // Start sending heartbeat messages
+  // Start sending ping messages
   //   interval - heartbeat interval in milliseconds
   //
   startHeartbeat(interval) {
-    const callback = () => {
-      this.transport.send('{}');
-      this.setTimeout(interval, this._heartbeatCallbackInstance);
-
-      if (process.env.NODE_ENV !== 'production') {
-        this.emit('heartbeat', this._heartbeatMessage);
+    const heartbeat = () => {
+      if (this._closed) {
+        return;
       }
+      const pingId = this._nextMessageId;
+      this.ping(() => {
+        this.session.unbufferUpTo(pingId);
+      });
     };
 
-    this._heartbeatCallbackInstance = callback;
-    callback();
+    this._heartbeatInterval = setInterval(heartbeat, interval);
   }
 
-  // Stop sending heartbeat messages
+  // Stop sending ping messages
   //
   stopHeartbeat() {
-    if (this._heartbeatCallbackInstance) {
-      this.clearTimeout(this._heartbeatCallbackInstance);
-      this._heartbeatCallbackInstance = null;
+    if (this._heartbeatInterval) {
+      clearTimeout(this._heartbeatInterval);
+      this._heartbeatInterval = null;
     }
   }
 
@@ -320,40 +319,17 @@ class Connection extends EventEmitter {
   // Close the connection
   //
   close() {
+    this._closed = true;
     this.stopHeartbeat();
     this.transport.end();
   }
 
-  // Set a timeout using timers.enroll()
-  //   milliseconds - amount of milliseconds
-  //   callback - callback function
-  //
-  setTimeout(milliseconds, callback) {
-    timers.enroll(this, milliseconds);
-    timers._unrefActive(this);
-    this.once('_timeout', callback);
-  }
-
-  // Clear a timeout set with Connection#setTimeout
-  //   handler - timer callback to remove
-  //
-  clearTimeout(handler) {
-    timers.unenroll(this);
-    this.removeListener('_timeout', handler);
-  }
-
   // Returns underlying transport
   //
   getTransport() {
     return this.transport.getRawTransport();
   }
 
-  // timers.enroll() timeout handler
-  //
-  _onTimeout() {
-    this.emit('_timeout');
-  }
-
   // Prepare a JSTP message to be sent over this connection
   //   message - a message to prepare
   //
@@ -383,6 +359,7 @@ class Connection extends EventEmitter {
   //   message - a message to send (optional)
   //
   _end(message) {
+    this._closed = true;
     this.stopHeartbeat();
 
     if (message) {
@@ -400,6 +377,7 @@ class Connection extends EventEmitter {
   // Closed socket event handler
   //
   _onSocketClose() {
+    this.closed = true;
     this.stopHeartbeat();
     this.emit('close', this);
     if (this.server) {

lib/server.js

@@ -161,10 +161,10 @@ class Server {
       }
     };
 
-    connection.setTimeout(HANDSHAKE_TIMEOUT, handleTimeout);
+    const handshakeTimeout = setTimeout(handleTimeout, HANDSHAKE_TIMEOUT);
 
     connection.on('client', () => {
-      connection.clearTimeout(handleTimeout);
+      clearTimeout(handshakeTimeout);
       if (this.heartbeatInterval) {
         connection.startHeartbeat(this.heartbeatInterval);
       }

lib/session.js

@@ -26,6 +26,12 @@ class Session extends Map {
     Object.preventExtensions(this);
   }
 
+  unbufferUpTo(id) {
+    for (let i = this.guaranteedDeliveredCount + 1; i <= id; i++) {
+      this.buffer.delete(i);
+    }
+  }
+
   _bufferMessage(id, message) {
     this.buffer.set(Math.abs(id), message);
     this.latestBufferedMessageId = id;
@@ -47,10 +53,8 @@ class Session extends Map {
     if (this.connection) {
       this.connection.close();
     }
+    this.unbufferUpTo(recievedCount);
     this.connection = newConnection;
-    for (let i = this.guaranteedDeliveredCount + 1; i <= recievedCount; i++) {
-      this.buffer.delete(i);
-    }
     this.guaranteedDeliveredCount = recievedCount;
   }
 

test/node/connection-emit-actions.js

@@ -7,10 +7,13 @@ const jstp = require('../..');
 
 const app = require('../fixtures/application');
 
+const HEARTBEAT_INTERVAL = 100;
+
 const application = new jstp.Application(app.name, app.interfaces);
 const serverConfig = {
   applications: [application],
   authPolicy: app.authCallback,
+  heartbeatInterval: HEARTBEAT_INTERVAL,
 };
 
 let server;
@@ -191,14 +194,34 @@ test.test('must emit messages in development mode', (test) => {
 });
 
 test.test('must emit heartbeat messages in development mode', (test) => {
-  test.plan(2);
+  test.plan(4);
+  const received = {
+    serverPing: false,
+    serverPong: false,
+    clientPing: false,
+    clientPong: false,
+  };
 
-  server.getClientsArray()[0].on('heartbeat', (message) => {
-    test.strictSame(message, {}, 'heartbeat message must match on server side');
+  server.getClientsArray()[0].on('incomingMessage', (message) => {
+    if (message.ping !== undefined) {
+      received.serverPing = true;
+    } else if (message.pong !== undefined) {
+      received.serverPong = true;
+    }
   });
-  connection.on('heartbeat', (message) => {
-    test.strictSame(message, {}, 'heartbeat message must match on client side');
+  connection.on('incomingMessage', (message) => {
+    if (message.ping !== undefined) {
+      received.clientPing = true;
+    } else if (message.pong !== undefined) {
+      received.clientPong = true;
+    }
   });
 
-  connection.startHeartbeat(100);
+  connection.startHeartbeat(HEARTBEAT_INTERVAL);
+  setTimeout(() => {
+    test.assert(received.serverPing, 'server must receive ping message');
+    test.assert(received.serverPong, 'server must receive pong message');
+    test.assert(received.clientPing, 'client must receive ping message');
+    test.assert(received.clientPing, 'client must receive pong message');
+  }, 2 * HEARTBEAT_INTERVAL);
 });

test/node/regress-gh-283.js

@@ -23,8 +23,10 @@ server.listen(() => {
 
       let heartbeatsCount = 0;
 
-      connection.on('heartbeat', () => {
-        heartbeatsCount++;
+      connection.on('incomingMessage', (message) => {
+        if (message.ping !== undefined) {
+          heartbeatsCount++;
+        }
       });
 
       setTimeout(() => {