Merge branch 'master' into breaking-changes

Author: mikuso

lib/client.js

@@ -1,6 +1,7 @@
 const { randomUUID} = require('crypto');
 const { EventEmitter, once } = require('events');
 const { setTimeout } = require('timers/promises');
+const { setTimeout: setTimeoutCb } = require('timers');
 const WebSocket = require('ws');
 const { ExponentialStrategy } = require('backoff');
 const { CONNECTING, OPEN, CLOSING, CLOSED } = WebSocket;
@@ -30,7 +31,6 @@ class RPCClient extends EventEmitter {
         this._keepAliveAbortController = undefined;
         this._pendingPingResponse = false;
         this._lastPingTime = 0;
-        this._skipNextPing = false;
         this._closePromise = undefined;
         this._protocolOptions = [];
         this._protocol = undefined;
@@ -407,12 +407,12 @@ class RPCClient extends EventEmitter {
         ws.on('error', err => this.emit('socketError', err));
         ws.on('ping', () => {
             if (this._options.deferPingsOnActivity) {
-                this._skipNextPing = true;
+                this._deferNextPing();
             }
         });
         ws.on('pong', () => {
             if (this._options.deferPingsOnActivity) {
-                this._skipNextPing = true;
+                this._deferNextPing();
             }
             this._pendingPingResponse = false;
             const rtt = Date.now() - this._lastPingTime;
@@ -566,10 +566,24 @@ class RPCClient extends EventEmitter {
         return this._connectPromise;
     }
 
+    _deferNextPing() {
+        if (!this._nextPingTimeout) {
+            return;
+        }
+
+        this._nextPingTimeout.refresh();
+    }
+
     async _keepAlive() {
         // abort any previously running keepAlive
         this._keepAliveAbortController?.abort();
 
+        const timerEmitter = new EventEmitter();
+        const nextPingTimeout = setTimeoutCb(()=>{
+            timerEmitter.emit('next')
+        }, this._options.pingIntervalMs);
+        this._nextPingTimeout = nextPingTimeout;
+
         try {
             if (this._state !== OPEN) {
                 // don't start pinging if connection not open
@@ -583,20 +597,16 @@ class RPCClient extends EventEmitter {
 
             // setup new abort controller
             this._keepAliveAbortController = new AbortController();
-
+            
             while (true) {
-                await setTimeout(this._options.pingIntervalMs, undefined, {signal: this._keepAliveAbortController.signal});
-                
+                await once(timerEmitter, 'next', {signal: this._keepAliveAbortController.signal}),
+                this._keepAliveAbortController.signal.throwIfAborted();
+
                 if (this._state !== OPEN) {
                     // keepalive no longer required
                     break;
                 }
 
-                if (this._skipNextPing) {
-                    this._skipNextPing = false;
-                    continue;
-                }
-
                 if (this._pendingPingResponse) {
                     // we didn't get a response to our last ping
                     throw Error("Ping timeout");
@@ -605,13 +615,17 @@ class RPCClient extends EventEmitter {
                 this._lastPingTime = Date.now();
                 this._pendingPingResponse = true;                
                 this._ws.ping();
+                nextPingTimeout.refresh();
             }
 
         } catch (err) {
+            // console.log('keepalive failed', err);
             if (err.name !== 'AbortError') {
                 // throws on ws.ping() error
                 this._ws.terminate();
             }
+        } finally {
+            clearTimeout(nextPingTimeout);
         }
     }
 
@@ -657,7 +671,7 @@ class RPCClient extends EventEmitter {
 
     _onMessage(buffer) {
         if (this._options.deferPingsOnActivity) {
-            this._skipNextPing = true;
+            this._deferNextPing();
         }
 
         const message = buffer.toString('utf8');

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "ocpp-rpc",
-  "version": "1.10.2",
+  "version": "1.10.3",
   "description": "A client & server implementation of the WAMP-like RPC-over-websocket system defined in the OCPP protcols (e.g. OCPP1.6-J and OCPP2.0.1).",
   "main": "index.js",
   "scripts": {

package.json

@@ -2479,8 +2479,8 @@ describe('RPCClient', function(){
                 withClient: async (client) => {
                     // send some rapid activity from the server
                     for (let i = 0; i < 4; i++) {
-                        await client.call('Echo', {});
                         await setTimeout(25);
+                        await client.call('Echo', {});
                     }
                     await client.close();
                 }
@@ -2490,7 +2490,7 @@ describe('RPCClient', function(){
                 identity: 'X',
                 reconnect: false,
                 deferPingsOnActivity: true,
-                pingIntervalMs: 40
+                pingIntervalMs: 50
             });
             cli.handle('Echo', async ({params}) => {
                 return params;
@@ -2517,8 +2517,8 @@ describe('RPCClient', function(){
                 withClient: async (client) => {
                     // send some rapid activity from the server
                     for (let i = 0; i < 4; i++) {
-                        await client.call('Echo', {});
                         await setTimeout(25);
+                        await client.call('Echo', {});
                     }
                     await client.close();
                 }
@@ -2528,7 +2528,7 @@ describe('RPCClient', function(){
                 identity: 'X',
                 reconnect: false,
                 deferPingsOnActivity: false,
-                pingIntervalMs: 40
+                pingIntervalMs: 50
             });
             cli.handle('Echo', async ({params}) => {
                 return params;
@@ -2567,7 +2567,7 @@ describe('RPCClient', function(){
                 identity: 'X',
                 reconnect: false,
                 deferPingsOnActivity: true,
-                pingIntervalMs: 40
+                pingIntervalMs: 50
             });
 
             // count how many times we ping the server
@@ -2602,7 +2602,7 @@ describe('RPCClient', function(){
                 identity: 'X',
                 reconnect: false,
                 deferPingsOnActivity: false,
-                pingIntervalMs: 40
+                pingIntervalMs: 50
             });
 
             // count how many times we ping the server