Merge pull request #273 from adpdigital/master

fixes #271
moscajs · Jul 12, 2015 · 18e7695 · 18e7695
2 parents ecfbe66 + 112676d
commit 18e7695
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 1 deletion.
diff --git a/lib/persistence/redis.js b/lib/persistence/redis.js
@@ -104,7 +104,9 @@ function RedisPersistence(options, callback) {
       }
 
       Object.keys(subs).forEach(function(sub) {
-        that._subMatcher.add(sub, id);
+        if (that._subMatcher.match(sub).indexOf(id) < 0) {
+          that._subMatcher.add(sub, id);
+        }
       });
 
       var redisError = null;

diff --git a/test/persistence/redis_spec.js b/test/persistence/redis_spec.js
@@ -189,6 +189,48 @@ describe("mosca.persistence.Redis", function() {
     });
 
   });
+
+
+  describe("clustered.environment", function(){
+
+    it("should forward each packet once after client reconnects", function(done) {
+      var client = {
+        id: "cluster client id - 42",
+        clean: false,
+        subscriptions: {
+          "hello/#": {
+            qos: 1
+          }
+        }
+      };
+
+      var packet = {
+        topic: "hello/42",
+        qos: 0,
+        payload: new Buffer("world"),
+        messageId: "42"
+      };
+
+      var that = this;
+      that.secondInstance = new Redis(opts, function() {
+        that.instance.storeSubscriptions(client, function() {
+          // simulate client reconnect since storeSubscriptions is called on disconnect
+          // no matter client connects to instance or secondInstance
+          that.secondInstance.storeSubscriptions(client, function () {
+            setTimeout(function () {
+              that.secondInstance.storeOfflinePacket(packet, function () {
+                that.instance.streamOfflinePackets(client, function (err, p) {
+                  expect(p).to.eql(packet);
+                  done(); // should be called once
+                });
+              });
+            }, 50);
+          });
+        });
+      });
+    });
+
+  });
 });