fix: keepalive issues

example.ts

@@ -1,24 +1,23 @@
 import mqtt from '.'
 
-const client = mqtt.connect('mqtt://test.mosquitto.org', {
-	keepalive: 10,
+const client = mqtt.connect('mqtts://test.mosquitto.org', {
+	keepalive: 60,
+	port: 8883,
 	reconnectPeriod: 15000,
+	rejectUnauthorized: false,
 })
 
 const testTopic = 'presence'
 
 function publish() {
-	client.publish(
-		testTopic,
-		`Hello mqtt ${new Date().toISOString()}`,
-		(err2) => {
-			if (!err2) {
-				console.log('message published')
-			} else {
-				console.error(err2)
-			}
-		},
-	)
+	const msg = `Hello mqtt ${new Date().toISOString()}`
+	client.publish(testTopic, msg, { qos: 1 }, (err2) => {
+		if (!err2) {
+			console.log('message published')
+		} else {
+			console.error(err2)
+		}
+	})
 }
 
 client.subscribe(testTopic, (err) => {
@@ -31,11 +30,12 @@ client.subscribe(testTopic, (err) => {
 
 client.on('message', (topic, message) => {
 	console.log('received message "%s" from topic "%s"', message, topic)
-	setTimeout(() => {
-		publish()
-	}, 2000)
 })
 
+setInterval(() => {
+	publish()
+}, 2000)
+
 client.on('error', (err) => {
 	console.error(err)
 })

src/lib/client.ts

@@ -424,8 +424,6 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 
 	public messageIdProvider: IMessageIdProvider
 
-	public pingResp: boolean
-
 	public outgoing: Record<
 		number,
 		{ volatile: boolean; cb: (err: Error, packet?: Packet) => void }
@@ -435,6 +433,9 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 
 	public noop: (error?: any) => void
 
+	/** Timestamp of last received control packet */
+	public pingResp: number
+
 	public pingTimer: PingTimer
 
 	/**
@@ -659,11 +660,7 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 			this.log('close :: clearing connackTimer')
 			clearTimeout(this.connackTimer)
 
-			this.log('close :: destroy ping timer')
-			if (this.pingTimer) {
-				this.pingTimer.destroy()
-				this.pingTimer = null
-			}
+			this._destroyPingTimer()
 
 			if (this.topicAliasRecv) {
 				this.topicAliasRecv.clear()
@@ -722,6 +719,7 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 	public connect() {
 		const writable = new Writable()
 		const parser = mqttPacket.parser(this.options)
+
 		let completeParse = null
 		const packets = []
 
@@ -1782,11 +1780,7 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 			this._setupReconnect()
 		}
 
-		if (this.pingTimer) {
-			this.log('_cleanUp :: destroy pingTimer')
-			this.pingTimer.destroy()
-			this.pingTimer = null
-		}
+		this._destroyPingTimer()
 
 		if (done && !this.connected) {
 			this.log(
@@ -1924,9 +1918,6 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 
 		this.emit('packetsend', packet)
 
-		// When writing a packet, reschedule the ping timer
-		this._shiftPingInterval()
-
 		this.log('_writePacket :: writing to stream')
 		const result = mqttPacket.writeToStream(
 			packet,
@@ -2084,18 +2075,27 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 		)
 
 		if (!this.pingTimer && this.options.keepalive) {
-			this.pingResp = true
 			this.pingTimer = new PingTimer(
 				this.options.keepalive,
 				() => {
 					this._checkPing()
 				},
 				this.options.timerVariant,
 			)
+			this.pingResp = Date.now()
+		}
+	}
+
+	private _destroyPingTimer() {
+		if (this.pingTimer) {
+			this.log('_destroyPingTimer :: destroying ping timer')
+			this.pingTimer.destroy()
+			this.pingTimer = null
 		}
 	}
 
 	/**
+
 	 * _shiftPingInterval - reschedule the ping interval
 	 *
 	 * @api private
@@ -2117,12 +2117,11 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 	 */
 	private _checkPing() {
 		this.log('_checkPing :: checking ping...')
-		if (this.pingResp) {
-			this.log(
-				'_checkPing :: ping response received. Clearing flag and sending `pingreq`',
-			)
-			this.pingResp = false
-			this._sendPacket({ cmd: 'pingreq' })
+		// 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'))
@@ -2131,6 +2130,11 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 		}
 	}
 
+	private _sendPing() {
+		this.log('_sendPing :: sending pingreq')
+		this._sendPacket({ cmd: 'pingreq' })
+	}
+
 	/**
 	 * _resubscribe
 	 * @api private

src/lib/handlers/index.ts

@@ -21,9 +21,18 @@ const handle: PacketHandler = (client, packet, done) => {
 		})
 		return client
 	}
+
 	client.log('_handlePacket :: emitting packetreceive')
 	client.emit('packetreceive', packet)
 
+	// 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 (packet.cmd !== 'pingresp') {
+		client['_shiftPingInterval']()
+	}
+
 	switch (packet.cmd) {
 		case 'publish':
 			handlePublish(client, packet, done)
@@ -49,7 +58,6 @@ const handle: PacketHandler = (client, packet, done) => {
 			break
 		case 'pingresp':
 			// this will be checked in _checkPing client method every keepalive interval
-			client.pingResp = true
 			done()
 			break
 		case 'disconnect':

test/abstract_client.ts

@@ -93,11 +93,13 @@ export default function abstractTest(server, config, ports) {
 
 			client.once('close', () => {
 				assert.notExists(client.pingTimer)
+
 				client.end(true, (err) => done(err))
 			})
 
 			client.once('connect', () => {
 				assert.exists(client.pingTimer)
+
 				client.stream.end()
 			})
 		})
@@ -1994,7 +1996,7 @@ export default function abstractTest(server, config, ports) {
 			})
 		})
 
-		it('should not checkPing if publishing at a higher rate than keepalive', function _test(t, done) {
+		it('should not shift ping on publish', function _test(t, done) {
 			const intervalMs = 3000
 			const client = connect({ keepalive: intervalMs / 1000 })
 
@@ -2007,7 +2009,7 @@ export default function abstractTest(server, config, ports) {
 				client.publish('foo', 'bar')
 				clock.tick(2)
 
-				assert.strictEqual(spy.callCount, 0)
+				assert.strictEqual(spy.callCount, 1)
 				client.end(true, done)
 			})
 		})
@@ -2067,13 +2069,16 @@ export default function abstractTest(server, config, ports) {
 					}
 				})
 
-				let client = connect({
+				const options: IClientOptions = {
 					keepalive: 60,
 					reconnectPeriod: 5000,
-				})
+				}
+
+				let client = connect()
 
 				client.once('connect', () => {
-					client.pingResp = false
+					// when using fake timers Date.now() counts from 0: https://sinonjs.org/releases/latest/fake-timers/
+					client.pingResp = -options.keepalive * 1000
 
 					client.once('error', (err) => {
 						assert.equal(err.message, 'Keepalive timeout')