mqtt - npm Package Compare versions

Comparing version 5.5.6 to 5.6.0

build/lib/client.d.ts

@@ -11,3 +11,3 @@ /// <reference types="node" />
 import { TypedEventEmitter } from './TypedEmitter';
-import PingTimer from './PingTimer';
+import KeepaliveManager from './KeepaliveManager';
 export type MqttProtocol = 'wss' | 'ws' | 'mqtt' | 'mqtts' | 'tcp' | 'ssl' | 'wx' | 'wxs' | 'ali' | 'alis';
@@ -149,4 +149,3 @@ export type StorePutCallback = () => void;
     noop: (error?: any) => void;
-    pingResp: number;
-    pingTimer: PingTimer;
+    keepaliveManager: KeepaliveManager;
     stream: IStream;
@@ -218,8 +217,8 @@ queue: {
     private _storePacket;
-    private _setupPingTimer;
-    private _destroyPingTimer;
-    private _shiftPingInterval;
+    private _setupKeepaliveManager;
+    private _destroyKeepaliveManager;
+    reschedulePing(): void;
     private _reschedulePing;
-    private _checkPing;
-    private _sendPing;
+    sendPing(): void;
+    onKeepaliveTimeout(): void;
     private _resubscribe;
@@ -226,0 +225,0 @@ private _onConnect;

build/lib/client.js

@@ -40,3 +40,3 @@ "use strict";
 const TypedEmitter_1 = require("./TypedEmitter");
-const PingTimer_1 = __importDefault(require("./PingTimer"));
+const KeepaliveManager_1 = __importDefault(require("./KeepaliveManager"));
 const is_browser_1 = __importStar(require("./is-browser"));
@@ -120,3 +120,3 @@ const setImmediate = globalThis.setImmediate ||
         this.messageIdToTopic = {};
-        this.pingTimer = null;
+        this.keepaliveManager = null;
         this.connected = false;
@@ -180,3 +180,3 @@ this.disconnecting = false;
             clearTimeout(this.connackTimer);
-            this._destroyPingTimer();
+            this._destroyKeepaliveManager();
             if (this.topicAliasRecv) {
@@ -843,3 +843,3 @@ this.topicAliasRecv.clear();
         }
-        this._destroyPingTimer();
+        this._destroyKeepaliveManager();
         if (done && !this.connected) {
@@ -1014,20 +1014,17 @@ this.log('_cleanUp :: (%s) :: removing stream `done` callback `close` listener', this.options.clientId);
     }
-    _setupPingTimer() {
-        this.log('_setupPingTimer :: keepalive %d (seconds)', this.options.keepalive);
-        if (!this.pingTimer && this.options.keepalive) {
-            this.pingTimer = new PingTimer_1.default(this.options.keepalive, () => {
-                this._checkPing();
-            }, this.options.timerVariant);
-            this.pingResp = Date.now();
+    _setupKeepaliveManager() {
+        this.log('_setupKeepaliveManager :: keepalive %d (seconds)', this.options.keepalive);
+        if (!this.keepaliveManager && this.options.keepalive) {
+            this.keepaliveManager = new KeepaliveManager_1.default(this, this.options.timerVariant);
         }
     }
-    _destroyPingTimer() {
-        if (this.pingTimer) {
-            this.log('_destroyPingTimer :: destroying ping timer');
-            this.pingTimer.destroy();
-            this.pingTimer = null;
+    _destroyKeepaliveManager() {
+        if (this.keepaliveManager) {
+            this.log('_destroyKeepaliveManager :: destroying keepalive manager');
+            this.keepaliveManager.destroy();
+            this.keepaliveManager = null;
         }
     }
-    _shiftPingInterval() {
-        if (this.pingTimer &&
+    reschedulePing() {
+        if (this.keepaliveManager &&
             this.options.keepalive &&
@@ -1040,21 +1037,13 @@ this.options.reschedulePings) {
         this.log('_reschedulePing :: rescheduling ping');
-        this.pingTimer.reschedule();
+        this.keepaliveManager.reschedule();
     }
-    _checkPing() {
-        this.log('_checkPing :: checking ping...');
-        const timeSincePing = Date.now() - this.pingResp - 100;
-        if (timeSincePing <= this.options.keepalive * 1000) {
-            this.log('_checkPing :: ping response received in time');
-            this._sendPing();
-        }
-        else {
-            this.emit('error', new Error('Keepalive timeout'));
-            this.log('_checkPing :: calling _cleanUp with force true');
-            this._cleanUp(true);
-        }
-    }
-    _sendPing() {
+    sendPing() {
         this.log('_sendPing :: sending pingreq');
         this._sendPacket({ cmd: 'pingreq' });
     }
+    onKeepaliveTimeout() {
+        this.emit('error', new Error('Keepalive timeout'));
+        this.log('onKeepaliveTimeout :: calling _cleanUp with force true');
+        this._cleanUp(true);
+    }
     _resubscribe() {
@@ -1100,3 +1089,3 @@ this.log('_resubscribe');
         this.messageIdProvider.clear();
-        this._setupPingTimer();
+        this._setupKeepaliveManager();
         this.connected = true;
@@ -1103,0 +1092,0 @@ const startStreamProcess = () => {

build/lib/get-timer.d.ts

@@ -1,8 +0,8 @@
-import { clearTimeout as clearT, setTimeout as setT } from 'worker-timers';
+import { clearInterval as clearI, setInterval as setI } from 'worker-timers';
 import type { TimerVariant } from './shared';
 export interface Timer {
-    set: typeof setT;
-    clear: typeof clearT;
+    set: typeof setI;
+    clear: typeof clearI;
 }
 declare const getTimer: (variant: TimerVariant) => Timer;
 export default getTimer;

build/lib/get-timer.js

@@ -29,8 +29,8 @@ "use strict";
 const workerTimer = {
-    set: worker_timers_1.setTimeout,
-    clear: worker_timers_1.clearTimeout,
+    set: worker_timers_1.setInterval,
+    clear: worker_timers_1.clearInterval,
 };
 const nativeTimer = {
-    set: (func, time) => setTimeout(func, time),
-    clear: (timerId) => clearTimeout(timerId),
+    set: (func, time) => setInterval(func, time),
+    clear: (timerId) => clearInterval(timerId),
 };
@@ -37,0 +37,0 @@ const getTimer = (variant) => {

build/lib/handlers/connack.js

@@ -28,3 +28,2 @@ "use strict";
             options.keepalive = packet.properties.serverKeepAlive;
-            client['_shiftPingInterval']();
         }
@@ -31,0 +30,0 @@ if (packet.properties.maximumPacketSize) {

build/lib/handlers/index.js

@@ -24,6 +24,2 @@ "use strict";
     }
-    client.pingResp = Date.now();
-    if (!['pingresp', 'publish'].includes(packet.cmd)) {
-        client['_shiftPingInterval']();
-    }
     client.log('_handlePacket :: emitting packetreceive');
@@ -40,2 +36,3 @@ client.emit('packetreceive', packet);
         case 'unsuback':
+            client.reschedulePing();
             (0, ack_1.default)(client, packet);
@@ -45,2 +42,3 @@ done();
         case 'pubrel':
+            client.reschedulePing();
             (0, pubrel_1.default)(client, packet, done);
@@ -53,2 +51,3 @@ break;
         case 'auth':
+            client.reschedulePing();
             (0, auth_1.default)(client, packet);
@@ -58,2 +57,4 @@ done();
         case 'pingresp':
+            client.log('_handlePacket :: received pingresp');
+            client.reschedulePing();
             done();
@@ -60,0 +61,0 @@ break;

build/mqtt.d.ts

@@ -6,7 +6,7 @@ import MqttClient from './lib/client';
 import connect, { connectAsync } from './lib/connect';
-import PingTimer from './lib/PingTimer';
+import KeepaliveManager from './lib/KeepaliveManager';
 export declare const Client: typeof MqttClient;
-export { connect, connectAsync, MqttClient, Store, DefaultMessageIdProvider, UniqueMessageIdProvider, IStore, PingTimer, };
+export { connect, connectAsync, MqttClient, Store, DefaultMessageIdProvider, UniqueMessageIdProvider, IStore, KeepaliveManager, };
 export * from './lib/client';
 export * from './lib/shared';
 export { ReasonCodes } from './lib/handlers/ack';

build/mqtt.js

@@ -32,3 +32,3 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.ReasonCodes = exports.PingTimer = exports.UniqueMessageIdProvider = exports.DefaultMessageIdProvider = exports.Store = exports.MqttClient = exports.connectAsync = exports.connect = exports.Client = void 0;
+exports.ReasonCodes = exports.KeepaliveManager = exports.UniqueMessageIdProvider = exports.DefaultMessageIdProvider = exports.Store = exports.MqttClient = exports.connectAsync = exports.connect = exports.Client = void 0;
 const client_1 = __importDefault(require("./lib/client"));
@@ -45,4 +45,4 @@ exports.MqttClient = client_1.default;
 Object.defineProperty(exports, "connectAsync", { enumerable: true, get: function () { return connect_1.connectAsync; } });
-const PingTimer_1 = __importDefault(require("./lib/PingTimer"));
-exports.PingTimer = PingTimer_1.default;
+const KeepaliveManager_1 = __importDefault(require("./lib/KeepaliveManager"));
+exports.KeepaliveManager = KeepaliveManager_1.default;
 exports.Client = client_1.default;
@@ -49,0 +49,0 @@ __exportStar(require("./lib/client"), exports);

package.json

 {
   "name": "mqtt",
   "description": "A library for the MQTT protocol",
-  "version": "5.5.6",
+  "version": "5.6.0",
   "contributors": [
@@ -6,0 +6,0 @@ "Adam Rudd <[e-mail]>",

src/lib/client.ts

@@ -42,3 +42,3 @@ /**
 import { TypedEventEmitter } from './TypedEmitter'
-import PingTimer from './PingTimer'
+import KeepaliveManager from './KeepaliveManager'
 import isBrowser, { isWebWorker } from './is-browser'
@@ -437,7 +437,4 @@
 
-	/** Timestamp of last received control packet */
-	public pingResp: number
+	public keepaliveManager: KeepaliveManager
 
-	public pingTimer: PingTimer
-
 	/**
@@ -577,4 +574,4 @@ * The connection to the Broker. In browsers env this also have `socket` property
 
-		// Ping timer, setup in _setupPingTimer
-		this.pingTimer = null
+		// Keepalive manager, setup in _setupKeepaliveManager
+		this.keepaliveManager = null
 		// Is the client connected?
@@ -666,3 +663,3 @@ this.connected = false
 
-			this._destroyPingTimer()
+			this._destroyKeepaliveManager()
 
@@ -1787,3 +1784,3 @@ if (this.topicAliasRecv) {
 
-		this._destroyPingTimer()
+		this._destroyKeepaliveManager()
 
@@ -2072,29 +2069,23 @@ if (done && !this.connected) {
 	/**
-	 * _setupPingTimer - setup the ping timer
-	 *
-	 * @api private
+	 * _setupKeepaliveManager - setup the keepalive manager
 	 */
-	private _setupPingTimer() {
+	private _setupKeepaliveManager() {
 		this.log(
-			'_setupPingTimer :: keepalive %d (seconds)',
+			'_setupKeepaliveManager :: keepalive %d (seconds)',
 			this.options.keepalive,
 		)
 
-		if (!this.pingTimer && this.options.keepalive) {
-			this.pingTimer = new PingTimer(
-				this.options.keepalive,
-				() => {
-					this._checkPing()
-				},
+		if (!this.keepaliveManager && this.options.keepalive) {
+			this.keepaliveManager = new KeepaliveManager(
+				this,
 				this.options.timerVariant,
 			)
-			this.pingResp = Date.now()
 		}
 	}
 
-	private _destroyPingTimer() {
-		if (this.pingTimer) {
-			this.log('_destroyPingTimer :: destroying ping timer')
-			this.pingTimer.destroy()
-			this.pingTimer = null
+	private _destroyKeepaliveManager() {
+		if (this.keepaliveManager) {
+			this.log('_destroyKeepaliveManager :: destroying keepalive manager')
+			this.keepaliveManager.destroy()
+			this.keepaliveManager = null
 		}
@@ -2104,10 +2095,7 @@ }
 	/**
-
-	 * _shiftPingInterval - reschedule the ping interval
-	 *
-	 * @api private
+	 * Reschedule the ping interval
 	 */
-	private _shiftPingInterval() {
+	public reschedulePing() {
 		if (
-			this.pingTimer &&
+			this.keepaliveManager &&
 			this.options.keepalive &&
@@ -2125,26 +2113,6 @@ this.options.reschedulePings
 		this.log('_reschedulePing :: rescheduling ping')
-		this.pingTimer.reschedule()
+		this.keepaliveManager.reschedule()
 	}
 
-	/**
-	 * _checkPing - check if a pingresp has come back, and ping the server again
-	 *
-	 * @api private
-	 */
-	private _checkPing() {
-		this.log('_checkPing :: checking ping...')
-		// give 100ms offset to avoid ping timeout when receiving fast responses
-		const timeSincePing = Date.now() - this.pingResp - 100
-		if (timeSincePing <= this.options.keepalive * 1000) {
-			this.log('_checkPing :: ping response received in time')
-			this._sendPing()
-		} else {
-			// do a forced cleanup since socket will be in bad shape
-			this.emit('error', new Error('Keepalive timeout'))
-			this.log('_checkPing :: calling _cleanUp with force true')
-			this._cleanUp(true)
-		}
-	}
-
-	private _sendPing() {
+	public sendPing() {
 		this.log('_sendPing :: sending pingreq')
@@ -2154,2 +2122,8 @@ this._sendPacket({ cmd: 'pingreq' })
 
+	public onKeepaliveTimeout() {
+		this.emit('error', new Error('Keepalive timeout'))
+		this.log('onKeepaliveTimeout :: calling _cleanUp with force true')
+		this._cleanUp(true)
+	}
+
 	/**
@@ -2217,3 +2191,3 @@ * _resubscribe
 		this.messageIdProvider.clear()
-		this._setupPingTimer()
+		this._setupKeepaliveManager()
 
@@ -2220,0 +2194,0 @@ this.connected = true

src/lib/get-timer.ts

 import isBrowser, { isWebWorker, isReactNativeBrowser } from './is-browser'
-import { clearTimeout as clearT, setTimeout as setT } from 'worker-timers'
+import { clearInterval as clearI, setInterval as setI } from 'worker-timers'
 import type { TimerVariant } from './shared'
@@ -9,14 +9,14 @@
 export interface Timer {
-	set: typeof setT
-	clear: typeof clearT
+	set: typeof setI
+	clear: typeof clearI
 }
 
 const workerTimer: Timer = {
-	set: setT,
-	clear: clearT,
+	set: setI,
+	clear: clearI,
 }
 
 const nativeTimer: Timer = {
-	set: (func, time) => setTimeout(func, time),
-	clear: (timerId) => clearTimeout(timerId),
+	set: (func, time) => setInterval(func, time),
+	clear: (timerId) => clearInterval(timerId),
 }
@@ -23,0 +23,0 @@

src/lib/handlers/connack.ts

@@ -32,4 +32,4 @@ import { ReasonCodes } from './ack'
 			options.keepalive = packet.properties.serverKeepAlive
-			client['_shiftPingInterval']()
 		}
+
 		if (packet.properties.maximumPacketSize) {
@@ -36,0 +36,0 @@ if (!options.properties) {

src/lib/handlers/index.ts

@@ -25,10 +25,2 @@ import handlePublish from './publish'
 
-	// keep track of last time we received a packet (for keepalive mechanism)
-	client.pingResp = Date.now()
-
-	// do not shift on pingresp otherwise we would skip the pingreq sending
-	if (!['pingresp', 'publish'].includes(packet.cmd)) {
-		client['_shiftPingInterval']()
-	}
-
 	client.log('_handlePacket :: emitting packetreceive')
@@ -39,2 +31,3 @@ client.emit('packetreceive', packet)
 		case 'publish':
+			// DO NOT SHIFT PING HERE, this would lead to https://github.com/mqttjs/MQTT.js/issues/1861
 			handlePublish(client, packet, done)
@@ -47,2 +40,3 @@ break
 		case 'unsuback':
+			client.reschedulePing()
 			handleAck(client, packet)
@@ -52,5 +46,8 @@ done()
 		case 'pubrel':
+			client.reschedulePing()
 			handlePubrel(client, packet, done)
 			break
 		case 'connack':
+			// no need to reschedule ping here as keepalive manager is created after successll connect
+			// (when onConnect is called at the end of handleConnack)
 			handleConnack(client, packet)
@@ -60,2 +57,3 @@ done()
 		case 'auth':
+			client.reschedulePing()
 			handleAuth(client, packet)
@@ -65,3 +63,4 @@ done()
 		case 'pingresp':
-			// this will be checked in _checkPing client method every keepalive interval
+			client.log('_handlePacket :: received pingresp')
+			client.reschedulePing()
 			done()
@@ -68,0 +67,0 @@ break

src/mqtt.ts

@@ -12,3 +12,3 @@ /*
 import connect, { connectAsync } from './lib/connect'
-import PingTimer from './lib/PingTimer'
+import KeepaliveManager from './lib/KeepaliveManager'
 
@@ -24,3 +24,3 @@ export const Client = MqttClient
 	IStore,
-	PingTimer,
+	KeepaliveManager,
 }
@@ -27,0 +27,0 @@ export * from './lib/client'

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

1847488

increased by0.23%

Lines of code

32224

increased by0.17%

No dependency changes