PUBLISH packet should NOT update the outgoing queue

.travis.yml

@@ -1,5 +1,6 @@
 language: node_js
 node_js:
-  - "6"
-  - "8"
-  - "10"
+  - 6
+  - 8
+  - 10
+  - 12

README.md

@@ -120,7 +120,7 @@ a client with `clean: false` connects to restore its subscriptions.
 
 -------------------------------------------------------
 <a name="countOffline"></a>
-### instance.countOffline(cb(err, numOfSubscriptions, numOfClients)
+### instance.countOffline(cb(err, numOfSubscriptions, numOfClients))
 
 Returns the number of offline subscriptions and the number of offline
 clients.

abstract.js

@@ -940,7 +940,7 @@ function abstractPersistence (opts) {
       instance.outgoingUpdate(client, updated, function (err, reclient, repacket) {
         t.error(err)
         t.equal(reclient, client, 'client matches')
-        t.equal(repacket, repacket, 'packet matches')
+        t.equal(repacket, updated, 'packet matches')
         callback(updated)
       })
     })
@@ -1027,6 +1027,55 @@ function abstractPersistence (opts) {
     })
   })
 
+  testInstance('update to publish w/ same messageId', function (t, instance) {
+    var sub = {
+      clientId: 'abcde', topic: 'hello', qos: 1
+    }
+    var client = {
+      id: sub.clientId
+    }
+    var packet1 = {
+      cmd: 'publish',
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 2,
+      dup: false,
+      length: 14,
+      retain: false,
+      brokerId: instance.broker.id,
+      brokerCounter: 42,
+      messageId: 42
+    }
+    var packet2 = {
+      cmd: 'publish',
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 2,
+      dup: false,
+      length: 14,
+      retain: false,
+      brokerId: instance.broker.id,
+      brokerCounter: 50,
+      messageId: 42
+    }
+
+    instance.outgoingEnqueue(sub, packet1, function () {
+      instance.outgoingEnqueue(sub, packet2, function () {
+        instance.outgoingUpdate(client, packet1, function () {
+          instance.outgoingUpdate(client, packet2, function () {
+            var stream = instance.outgoingStream(client)
+            stream.pipe(concat(function (list) {
+              t.equal(list.length, 2, 'must have two items in queue')
+              t.deepEqual(list[0].brokerCounter, packet1.brokerCounter, 'packet must match')
+              t.deepEqual(list[1].brokerCounter, packet2.brokerCounter, 'packet must match')
+              instance.destroy(t.end.bind(t))
+            }))
+          })
+        })
+      })
+    })
+  })
+
   testInstance('update to pubrel', function (t, instance) {
     var sub = {
       clientId: 'abcde', topic: 'hello', qos: 1
@@ -1054,7 +1103,7 @@ function abstractPersistence (opts) {
       instance.outgoingUpdate(client, updated, function (err, reclient, repacket) {
         t.error(err)
         t.equal(reclient, client, 'client matches')
-        t.equal(repacket, repacket, 'packet matches')
+        t.equal(repacket, updated, 'packet matches')
 
         var pubrel = {
           cmd: 'pubrel',
@@ -1217,7 +1266,7 @@ function abstractPersistence (opts) {
         t.error(err, 'no error')
         t.equal(c, anotherClient, 'client matches')
         instance.streamWill({
-          'anotherBroker': Date.now()
+          anotherBroker: Date.now()
         })
           .pipe(through.obj(function (chunk, enc, cb) {
             t.deepEqual(chunk, {

persistence.js

@@ -202,7 +202,8 @@ MemoryPersistence.prototype.outgoingUpdate = function (client, packet, cb) {
       temp.brokerCounter === packet.brokerCounter) {
       temp.messageId = packet.messageId
       return cb(null, client, packet)
-    } else if (temp.messageId === packet.messageId) {
+    } else if (packet.cmd !== 'publish' && temp.messageId === packet.messageId) {
+      // for non-PUBLISH packet only
       outgoing[i] = packet
       return cb(null, client, packet)
     }