Merge 5d35282 into e2d64b2

test/qos1.js

@@ -1,5 +1,6 @@
 'use strict'
 
+var concat = require('concat-stream')
 var Buffer = require('safe-buffer').Buffer
 var test = require('tape').test
 var helper = require('./helper')
@@ -273,12 +274,15 @@ test('remove stored subscriptions if connected with clean=true', function (t) {
 })
 
 test('resend publish on non-clean reconnect QoS 1', function (t) {
-  t.plan(6)
+  t.plan(8)
 
   var broker = aedes()
   var publisher
   var opts = { clean: false, clientId: 'abcde' }
   var subscriber = connect(setup(broker), opts)
+  var subscriberClient = {
+    id: opts.clientId
+  }
   var expected = {
     cmd: 'publish',
     topic: 'hello',
@@ -301,7 +305,13 @@ test('resend publish on non-clean reconnect QoS 1', function (t) {
       qos: 1,
       messageId: 42
     })
-
+    publisher.inStream.write({
+      cmd: 'publish',
+      topic: 'hello',
+      payload: 'world world',
+      qos: 1,
+      messageId: 42
+    })
     publisher.outStream.once('data', function (packet) {
       t.equal(packet.cmd, 'puback')
 
@@ -315,7 +325,13 @@ test('resend publish on non-clean reconnect QoS 1', function (t) {
         t.notEqual(packet.messageId, 42, 'messageId must differ')
         delete packet.messageId
         t.deepEqual(packet, expected, 'packet must match')
-        t.end()
+        setImmediate(() => {
+          var stream = broker.persistence.outgoingStream(subscriberClient)
+          stream.pipe(concat(function (list) {
+            t.equal(list.length, 1, 'should remain one item in queue')
+            t.deepEqual(list[0].payload, Buffer.from('world world'), 'packet must match')
+          }))
+        })
       })
     })
   })

test/qos2.js

@@ -425,3 +425,35 @@ test('publish after disconnection', function (t) {
     })
   })
 })
+
+test('multiple publish and store one', function (t) {
+  t.plan(2)
+
+  var broker = aedes()
+  var sid = {
+    id: 'abcde'
+  }
+  var s = connect(setup(broker), { clientId: sid.id })
+  var toPublish = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 2,
+    retain: false,
+    messageId: 42
+  }
+
+  var count = 5
+  while (--count) {
+    s.inStream.write(toPublish)
+  }
+  broker.on('closed', function () {
+    broker.persistence.incomingGetPacket(sid, toPublish, function (err, origPacket) {
+      delete origPacket.brokerId
+      delete origPacket.brokerCounter
+      t.deepEqual(origPacket, toPublish, 'packet must match')
+      t.error(err)
+      t.end()
+    })
+  })
+})