Queue packets if the client is disconnected

Adam Rudd · Adam Rudd · commit 20bec5974b3b · 2013-05-26T17:53:50.000+02:00
diff --git a/lib/client.js b/lib/client.js
@@ -60,6 +60,8 @@ function MqttClient(stream, options) {
   this.pingTimer = null;
   // Is the client connected?
   this.connected = false;
+  // Packet queue
+  this.queue = [];
 
   // Inflight messages
   this.inflight = {
@@ -88,9 +90,21 @@ function MqttClient(stream, options) {
   // Setup ping timer
   this.on('connect', this._setupPingTimer);
 
-  // Mark the client as connected
+  // Send queued packets
   this.on('connect', function() {
     that.connected = true;
+
+    var queue = that.queue
+      , length = queue.length;
+
+    for (var i = 0; i < length; i += 1) {
+      that._sendPacket(
+        queue[i].type,
+        queue[i].packet,
+        queue[i].cb
+      );
+    }
+    that.queue = [];
   });
 
   // Handle incoming publish
@@ -298,12 +312,20 @@ MqttClient.prototype.end = function() {
   }
 };
 
+MqttClient.prototype._sendPacket = function(type, packet, cb) {
+  if (this.connected) {
+    this.conn[type](packet);
+    if (cb) cb.call(this);
+  } else {
+    this.queue.push({type: type, packet: packet, cb: cb});
+  }
+};
+
 /**
  * _setupPingTimer
  * 
  * @api private
  */
-
 MqttClient.prototype._setupPingTimer = function() {
   // No ping
   if (this.options.keepalive === 0) {
diff --git a/test/client.js b/test/client.js
@@ -211,7 +211,23 @@ describe('MqttClient', function () {
   });
 
   describe('publishing', function() {
-    it('should publish a message', function (done) {
+    it('should queue message until connected', function(done) {
+      var client = createClient(port);
+
+      client.publish('test', 'test');
+      client.queue.length.should.equal(1);
+
+      client.once('connect', function() {
+        client.queue.length.should.equal(0);
+        done();
+      });
+
+      this.server.once('client', function(client) {
+        client.once('connect', function(packet) {
+          client.connack({returnCode: 0});
+        });
+      });
+    });
       var client = createClient(port)
         , payload = 'test'
         , topic = 'test';
