Merge pull request mqttjs#864 from redboltz/fix_store_pub_after_proc

mcollina · web-flow · commit adff8408ec51 · 2018-08-30T17:44:55.000+02:00
Added error handling to imcomingStore.put callback in _handlePublish.
diff --git a/lib/client.js b/lib/client.js
@@ -954,6 +954,7 @@ default:
 for now i just suppressed the warnings
 */
 MqttClient.prototype._handlePublish = function (packet, done) {
+  done = typeof done !== 'undefined' ? done : nop
   var topic = packet.topic.toString()
   var message = packet.payload
   var qos = packet.qos
@@ -962,7 +963,10 @@ MqttClient.prototype._handlePublish = function (packet, done) {
 
   switch (qos) {
     case 2:
-      this.incomingStore.put(packet, function () {
+      this.incomingStore.put(packet, function (err) {
+        if (err) {
+          return done(err)
+        }
         that._sendPacket({cmd: 'pubrec', messageId: mid}, done)
       })
       break
@@ -971,7 +975,7 @@ MqttClient.prototype._handlePublish = function (packet, done) {
       this.emit('message', topic, message, packet)
       this.handleMessage(packet, function (err) {
         if (err) {
-          return done && done(err)
+          return done(err)
         }
         // send 'puback' if the above 'handleMessage' method executed
         // successfully.
@@ -1074,6 +1078,7 @@ MqttClient.prototype._handleAck = function (packet) {
  * @api private
  */
 MqttClient.prototype._handlePubrel = function (packet, callback) {
+  callback = typeof callback !== 'undefined' ? callback : nop
   var mid = packet.messageId
   var that = this
 
@@ -1082,12 +1087,16 @@ MqttClient.prototype._handlePubrel = function (packet, callback) {
   that.incomingStore.get(packet, function (err, pub) {
     if (!err && pub.cmd !== 'pubrel') {
       that.emit('message', pub.topic, pub.payload, pub)
-      that.incomingStore.put(packet)
-      that.handleMessage(pub, function (err) {
+      that.incomingStore.put(packet, function (err) {
         if (err) {
-          return callback && callback(err)
+          return callback(err)
         }
-        that._sendPacket(comp, callback)
+        that.handleMessage(pub, function (err) {
+          if (err) {
+            return callback(err)
+          }
+          that._sendPacket(comp, callback)
+        })
       })
     } else {
       that._sendPacket(comp, callback)
diff --git a/test/abstract_client.js b/test/abstract_client.js
@@ -973,6 +973,115 @@ module.exports = function (server, config) {
       }
     })
 
+    it('should handle error with async incoming store in QoS 2 `handlePublish` method', function (done) {
+      function AsyncStore () {
+        if (!(this instanceof AsyncStore)) {
+          return new AsyncStore()
+        }
+      }
+      AsyncStore.prototype.put = function (packet, cb) {
+        process.nextTick(function () {
+          cb(new Error('Error'))
+        })
+      }
+      var store = new AsyncStore()
+      var client = connect({incomingStore: store})
+
+      client._handlePublish({
+        messageId: 1,
+        topic: 'test',
+        payload: 'test',
+        qos: 2
+      }, function () {
+        done()
+        client.end()
+      })
+    })
+
+    it('should handle error with async incoming store in QoS 2 `handlePubrel` method', function (done) {
+      function AsyncStore () {
+        if (!(this instanceof AsyncStore)) {
+          return new AsyncStore()
+        }
+      }
+      AsyncStore.prototype.put = function (packet, cb) {
+        process.nextTick(function () {
+          cb(new Error('Error'))
+        })
+      }
+      AsyncStore.prototype.get = function (packet, cb) {
+        process.nextTick(function () {
+          cb(null, {cmd: 'publish'})
+        })
+      }
+      var store = new AsyncStore()
+      var client = connect({incomingStore: store})
+
+      client._handlePubrel({
+        messageId: 1,
+        qos: 2
+      }, function () {
+        done()
+        client.end()
+      })
+    })
+
+    it('should handle success with async incoming store in QoS 2 `handlePubrel` method', function (done) {
+      var putComplete = false
+      function AsyncStore () {
+        if (!(this instanceof AsyncStore)) {
+          return new AsyncStore()
+        }
+      }
+      AsyncStore.prototype.put = function (packet, cb) {
+        process.nextTick(function () {
+          putComplete = true
+          cb(null)
+        })
+      }
+      AsyncStore.prototype.get = function (packet, cb) {
+        process.nextTick(function () {
+          cb(null, {cmd: 'publish'})
+        })
+      }
+      var store = new AsyncStore()
+      var client = connect({incomingStore: store})
+
+      client._handlePubrel({
+        messageId: 1,
+        qos: 2
+      }, function () {
+        putComplete.should.equal(true)
+        done()
+        client.end()
+      })
+    })
+
+    it('should handle error with async incoming store in QoS 1 `handlePublish` method', function (done) {
+      function AsyncStore () {
+        if (!(this instanceof AsyncStore)) {
+          return new AsyncStore()
+        }
+      }
+      AsyncStore.prototype.put = function (packet, cb) {
+        process.nextTick(function () {
+          cb(null, 'Error')
+        })
+      }
+      var store = new AsyncStore()
+      var client = connect({incomingStore: store})
+
+      client._handlePublish({
+        messageId: 1,
+        topic: 'test',
+        payload: 'test',
+        qos: 1
+      }, function () {
+        done()
+        client.end()
+      })
+    })
+
     it('should not send a `pubcomp` if the execution of `handleMessage` fails for messages with QoS `2`', function (done) {
       var store = new Store()
       var client = connect({incomingStore: store})
