Removed persistent subscription while online.

lib/persistence/levelup.js

@@ -141,11 +141,18 @@ LevelUpPersistence.prototype.lookupSubscriptions = function(client, done) {
   var that = this;
   this._clientSubscriptions.get(client.id, function(err, subscriptions) {
     that._clientSubscriptions.del(client.id, function() {
+      var levelkeys;
+
+      levelkeys = Object.keys(subscriptions || {}).map(function(key) {
+        // TODO we need to remove these from the subLobber every time.
+        var levelKey = util.format("%s:%s", key, client.id);
+        that._subLobber.remove(key, levelKey);
+        return levelKey;
+      });
+
       if (subscriptions && client.clean) {
         that.streamOfflinePackets(client, nop, function() {
-          that._subscriptions.batch(Object.keys(subscriptions).map(function(key) {
-            var levelKey = util.format("%s:%s", key, client.id);
-            that._subLobber.remove(key, levelKey);
+          that._subscriptions.batch(levelkeys.map(function(levelKey) {
             return {
               key: levelKey,
               type: 'del'
@@ -160,7 +167,8 @@ LevelUpPersistence.prototype.lookupSubscriptions = function(client, done) {
         if (done) {
           done(null, subscriptions);
         }
-      }
+        return;
+      } 
     });
   });
 };

lib/persistence/redis.js

@@ -243,10 +243,16 @@ RedisPersistence.prototype.lookupSubscriptions = function(client, cb) {
   var key = "client:sub:" + client.id;
   var subscriptions;
 
-  var multi = this._client.multi()
+  var multi = this._client.multi();
+  var that = this;
 
   multi.get(key, function(err, result) {
+
     subscriptions = JSON.parse(result) || {};
+
+    Object.keys(subscriptions).forEach(function(sub) {
+      that._subLobber.remove(sub, client.id);
+    });
   });
 
   multi.del(key);

test/persistence/abstract.js

@@ -612,6 +612,27 @@ module.exports = function(create, buildOpts) {
       });
     });
 
+    it("should not store any offline packet for a client after lookup", function(done) {
+      var instance = this.instance;
+      var client = {
+        id: "my client id - 42",
+        clean: false,
+        logger: globalLogger,
+        subscriptions: {
+          hello: 1
+        }
+      };
+
+      instance.lookupSubscriptions(client, function(err, results) {
+        instance.storeOfflinePacket(packet, function() {
+          instance.streamOfflinePackets(client, function(err, p) {
+            done(new Error("this should never be called"));
+          });
+          done();
+        });
+      });
+    });
+
     it("should not stream any offline packet to a clean client", function(done) {
       var instance = this.instance;
       var client = {