Added support for storing inflight packets for client.

moscajs · mcollina · Jun 26, 2013 · Jun 17, 2013 · Jun 17, 2013 · Jun 18, 2013
commit 722af400935494189743af8e7f886b6a863bd2fc
diff --git a/lib/persistance/abstract.js b/lib/persistance/abstract.js
@@ -1,5 +1,7 @@
 "use strict";
 
+var async = require("async");
+
 function AbstractPersistence() {
 
 }
@@ -60,7 +62,17 @@ AbstractPersistence.prototype.wire = function(server) {
 
   server.on("clientDisconnecting", function(client) {
     that.storeSubscriptions(client);
+    that.storeInflightPackets(client);
   });
 };
 
+AbstractPersistence.prototype.storeInflightPackets = function(client, done) {
+  if (client.inflight) {
+    var that = this;
+    async.each(Object.keys(client.inflight), function(key, cb) {
+      that._storePacket(client.id, client.inflight[key].packet, cb);
+    }, done);
+  }
+};
+
 module.exports = AbstractPersistence;
diff --git a/lib/persistance/levelup.js b/lib/persistance/levelup.js
@@ -127,18 +127,13 @@ LevelUpPersistance.prototype.lookupSubscriptions = function(client, done) {
 LevelUpPersistance.prototype.storeOfflinePacket = function(packet, done) {
   var that = this;
   var subs = this._subLobber.match(packet.topic);
-  var ttl = {
-    ttl: that.options.ttl.subscriptions
-  };
 
   async.each(subs, function(key, cb) {
     that._subscriptions.get(key, function(err, sub) {
       if (err) {
         return cb(err);
       }
-      var key = util.format("%s:%s", sub.client, new Date().toISOString());
-      that._offlinePackets.put(
-        key, packet, ttl, cb);
+      that._storePacket(sub.client, packet, cb);
     });
   }, done);
 };
@@ -165,6 +160,15 @@ LevelUpPersistance.prototype.streamOfflinePackets = function(client, cb, done) {
   }
 };
 
+LevelUpPersistance.prototype._storePacket = function(client, packet, cb) {
+  var key = util.format("%s:%s", client, new Date().toISOString());
+  var ttl = {
+    ttl: this.options.ttl.subscriptions
+  };
+  this._offlinePackets.put(
+    key, packet, ttl, cb);
+};
+
 LevelUpPersistance.prototype.close = function(cb) {
   this.db.close(cb);
 };

diff --git a/lib/persistance/mongo.js b/lib/persistance/mongo.js
@@ -160,10 +160,7 @@ MongoPersistance.prototype.storeOfflinePacket = function(packet, done) {
   stream.on("data", function(data) {
     started++;
 
-    that._packets.insert({
-      client: data.client,
-      packet: packet
-    }, function(err) {
+    that._storePacket(data.client, packet, function(err) {
       if (err) {
         return stream.emit("error", err);
       }
@@ -185,6 +182,13 @@ MongoPersistance.prototype.storeOfflinePacket = function(packet, done) {
   });
 };
 
+MongoPersistance.prototype._storePacket = function(client, packet, cb) {
+  this._packets.insert({
+    client: client,
+    packet: packet
+  }, cb);
+};
+
 MongoPersistance.prototype.streamOfflinePackets = function(client, cb) {
   if (client.clean) {
     return;

diff --git a/lib/persistance/redis.js b/lib/persistance/redis.js
@@ -177,10 +177,14 @@ RedisPersistance.prototype.storeOfflinePacket = function(packet, done) {
 
   var matches = this._subLobber.match(packet.topic);
   async.each(matches, function(client, cb) {
-    that._client.lpush("packets:" + client, JSON.stringify(packet), cb);
+    that._storePacket(client, packet, cb);
   }, done);
 };
 
+RedisPersistance.prototype._storePacket = function(client, packet, cb) {
+  this._client.lpush("packets:" + client, JSON.stringify(packet), cb);
+};
+
 RedisPersistance.prototype.streamOfflinePackets = function(client, cb) {
   var that = this;
 

diff --git a/test/persistance/abstract.js b/test/persistance/abstract.js
@@ -566,4 +566,65 @@ module.exports = function(create) {
       });
     });
   });
+
+  describe("inflight packets", function() {
+    var packet = {
+      topic: "hello",
+      qos: 0,
+      payload: "world",
+      messageId: 42
+    };
+    var client = { 
+      id: "my client id - 42",
+      clean: false,
+      subscriptions: {
+        hello: {
+          qos: 1
+        }
+      },
+      inflight: {
+        42: { packet: packet }
+      }
+    };
+
+    it("should store one inflight packet", function(done) {
+      this.instance.storeInflightPackets(client, done);
+    });
+
+    it("should store and stream an inflight packet", function(done) {
+      var instance = this.instance;
+      instance.storeInflightPackets(client, function() {
+        instance.streamOfflinePackets(client, function(err, p) {
+          expect(p).to.eql(packet);
+          done();
+        });
+      });
+    });
+
+    it("should delete the offline packets once streamed", function(done) {
+      var instance = this.instance;
+      instance.storeInflightPackets(client, function() {
+        instance.streamOfflinePackets(client, function(err, p) {
+          instance.streamOfflinePackets(client, function(err, p2) {
+            done(new Error("this should never be called"));
+          });
+          done();
+        });
+      });
+    });
+
+    it("should wire itself up to the 'clientDisconnecting' event of a Server", function(done) {
+      var em = new EventEmitter();
+      var instance = this.instance;
+      instance.wire(em);
+
+      em.emit("clientDisconnecting", client);
+
+      setTimeout(function() {
+        instance.streamOfflinePackets(client, function(err, packet) {
+          done();
+        });
+      }, 20); // 20ms will suffice 
+    });
+  });
 };