fix #1254

lib/client.js

@@ -1195,9 +1195,10 @@ MqttClient.prototype._cleanUp = function (forced, done) {
  * @param {Object} packet - packet options
  * @param {Function} cb - callback when the packet is sent
  * @param {Function} cbStorePut - called when message is put into outgoingStore
+ * @param {Boolean} noStore - send without put to the store
  * @api private
  */
-MqttClient.prototype._sendPacket = function (packet, cb, cbStorePut) {
+MqttClient.prototype._sendPacket = function (packet, cb, cbStorePut, noStore) {
   debug('_sendPacket :: (%s) ::  start', this.options.clientId)
   cbStorePut = cbStorePut || nop
   cb = cb || nop
@@ -1224,6 +1225,11 @@ MqttClient.prototype._sendPacket = function (packet, cb, cbStorePut) {
   // When sending a packet, reschedule the ping timer
   this._shiftPingInterval()
 
+  if (noStore) {
+    sendPacket(this, packet, cb)
+    return
+  }
+
   switch (packet.cmd) {
     case 'publish':
       break
@@ -1840,7 +1846,7 @@ MqttClient.prototype._onConnect = function (packet) {
         }
         that._packetIdsDuringStoreProcessing[packet.messageId] = true
         if (that.messageIdProvider.register(packet.messageId)) {
-          that._sendPacket(packet)
+          that._sendPacket(packet, undefined, undefined, true)
         } else {
           debug('messageId: %d has already used.', packet.messageId)
         }

package.json

@@ -103,7 +103,8 @@
     "standard": "^16.0.4",
     "tape": "^5.5.2",
     "terser": "^5.14.2",
-    "typescript": "^4.5.5"
+    "typescript": "^4.5.5",
+    "mqtt-level-store": "^3.1.0"
   },
   "standard": {
     "env": [

test/abstract_client.js

@@ -11,6 +11,8 @@ const Store = require('./../lib/store')
 const assert = require('chai').assert
 const ports = require('./helpers/port_list')
 const serverBuilder = require('./server_helpers_for_client_tests').serverBuilder
+const fs = require('fs')
+const levelStore = require('mqtt-level-store')
 
 module.exports = function (server, config) {
   const version = config.protocolVersion || 4
@@ -652,6 +654,77 @@ module.exports = function (server, config) {
       })
     })
 
+    it('should not overtake the messages stored in the level-db-store', function (done) {
+      var storePath = './temp'
+      var store = levelStore(storePath)
+      var client = null
+      var incomingStore = store.incoming
+      var outgoingStore = store.outgoing
+      var publishCount = 0
+
+      var server2 = serverBuilder(config.protocol, function (serverClient) {
+        serverClient.on('connect', function () {
+          var connack = version === 5 ? { reasonCode: 0 } : { returnCode: 0 }
+          serverClient.connack(connack)
+        })
+        serverClient.on('publish', function (packet) {
+          if (packet.qos !== 0) {
+            serverClient.puback({messageId: packet.messageId})
+          }
+
+          switch (publishCount++) {
+            case 0:
+              assert.strictEqual(packet.payload.toString(), 'payload1')
+              break
+            case 1:
+              assert.strictEqual(packet.payload.toString(), 'payload2')
+              break
+            case 2:
+              assert.strictEqual(packet.payload.toString(), 'payload3')
+
+              server2.close()
+              fs.rmdirSync(storePath, {recursive: true})
+              done()
+              break
+          }
+        })
+      })
+
+      const clientOptions = {
+        port: ports.PORTAND72,
+        host: 'localhost',
+        clean: false,
+        clientId: 'cid1',
+        reconnectPeriod: 0,
+        incomingStore: incomingStore,
+        outgoingStore: outgoingStore,
+        queueQoSZero: true
+      }
+
+      server2.listen(ports.PORTAND72, function () {
+        client = connect(clientOptions)
+
+        client.once('close', function () {
+          client.once('connect', function () {
+            client.publish('test', 'payload2', {qos: 1})
+            client.publish('test', 'payload3', {qos: 1}, function () {
+              client.end(false)
+            })
+          })
+          // reconecting
+          client.reconnect(clientOptions)
+        })
+
+        // publish and close
+        client.publish('test', 'payload1', {
+          qos: 1,
+          cbStorePut: function () {
+            client.end(true)
+          }
+        })
+      })
+    })
+
     it('should call cb if an outgoing QoS 0 message is not sent', function (done) {
       const client = connect({ queueQoSZero: false })
       let called = false