destroy the ongoing stream in case of a disconnect

mcollina · mcollina · commit 59fd16e5868e · 2018-08-25T14:48:47.000+02:00
diff --git a/lib/client.js b/lib/client.js
@@ -136,10 +136,28 @@ function MqttClient (streamBuilder, options) {
     }
 
     this.connected = true
-    var outStore = null
-    outStore = this.outgoingStore.createStream()
+    var outStore = this.outgoingStore.createStream()
+
+    this.once('close', remove)
+    outStore.on('end', function () {
+      that.removeListener('close', remove)
+    })
+    outStore.on('error', function (err) {
+      that.removeListener('close', remove)
+      that.emit('error', err)
+    })
+
+    function remove () {
+      outStore.destroy()
+      outStore = null
+    }
 
     function storeDeliver () {
+      // edge case, we wrapped this twice
+      if (!outStore) {
+        return
+      }
+
       var packet = outStore.read(1)
       var cb
 
@@ -166,9 +184,6 @@ function MqttClient (streamBuilder, options) {
       }
     }
 
-    // Control of stored messages
-    outStore.on('error', this.emit.bind(this, 'error'))
-
     // start flowing
     storeDeliver()
   })
diff --git a/test/abstract_store.js b/test/abstract_store.js
@@ -130,4 +130,30 @@ module.exports = function abstractStoreTest (build) {
       })
     })
   })
+
+  it('should replace a packet when doing put with the same messageId', function (done) {
+    var packet1 = {
+      cmd: 'publish', // added
+      topic: 'hello',
+      payload: 'world',
+      qos: 2,
+      messageId: 42
+    }
+    var packet2 = {
+      cmd: 'pubrel', // added
+      qos: 2,
+      messageId: 42
+    }
+
+    store.put(packet1, function () {
+      store.put(packet2, function () {
+        store
+          .createStream()
+          .on('data', function (data) {
+            data.should.eql(packet2)
+            done()
+          })
+      })
+    })
+  })
 }
