Merge pull request #399 from Densaugeo/master

Add .qosZeroNotQueued option to MqttClient

README.md

@@ -188,6 +188,7 @@ the `connect` event. Typically a `net.Socket`.
   * `password`: the password required by your broker, if any
   * `incomingStore`: a [Store](#store) for the incoming packets
   * `outgoingStore`: a [Store](#store) for the outgoing packets
+  * `queueQoSZero`: if connection is broken, queue outgoing QoS zero messages (default `true`)
   * `will`: a message that will sent by the broker automatically when
      the client disconnect badly. The format is:
     * `topic`: the topic to publish

lib/client.js

@@ -90,6 +90,9 @@ function MqttClient (streamBuilder, options) {
   this.outgoingStore = this.options.outgoingStore || new Store();
   this.incomingStore = this.options.incomingStore || new Store();
 
+  // Should QoS zero messages be queued when the connection is broken?
+  this.queueQoSZero = null == this.options.queueQoSZero ? true : this.options.queueQoSZero;
+
   // Ping timer, setup in _setupPingTimer
   this.pingTimer = null;
   // Is the client connected?
@@ -597,7 +600,13 @@ MqttClient.prototype._cleanUp = function (forced, done) {
  */
 MqttClient.prototype._sendPacket = function (packet, cb) {
   if (!this.connected) {
-    return this.queue.push({ packet: packet, cb: cb });
+    if (0 < packet.qos || 'publish' !== packet.cmd || this.queueQoSZero) {
+      this.queue.push({ packet: packet, cb: cb });
+    } else if (cb) {
+      cb(new Error('No connection to broker'));
+    }
+
+    return;
   }
 
   // When sending a packet, reschedule the ping timer

test/abstract_client.js

@@ -271,6 +271,31 @@ module.exports = function (server, config) {
       });
     });
 
+    it('should not queue qos 0 messages if queueQoSZero is false', function () {
+      var client = connect({queueQoSZero: false});
+
+      client.publish('test', 'test', {qos: 0});
+      client.queue.length.should.equal(0);
+    });
+
+    it('should still queue qos != 0 messages if queueQoSZero is false', function () {
+      var client = connect({queueQoSZero: false});
+
+      client.publish('test', 'test', {qos: 1});
+      client.publish('test', 'test', {qos: 2});
+      client.subscribe('test');
+      client.unsubscribe('test');
+      client.queue.length.should.equal(4);
+    });
+
+    it('should call cb if an outgoing QoS 0 message is not sent', function (done) {
+      var client = connect({queueQoSZero: false});
+
+      client.publish('test', 'test', {qos: 0}, function () {
+        done();
+      });
+    });
+
     if (!process.env.TRAVIS) {
       it('should queue message until connected', function (done) {
         var client = connect();