Merge a6ffa75 into c76b680

moscajs · Apr 14, 2016 · 17e209f · 17e209f
2 parents c76b680 + a6ffa75
commit 17e209f
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 16 deletions.
diff --git a/lib/persistence/levelup.js b/lib/persistence/levelup.js
@@ -149,7 +149,7 @@ LevelUpPersistence.prototype.lookupRetained = function(pattern, cb) {
   });
 
   stream.on("data", function(data) {
-    if (matcher.match(data.key).length > 0) {
+    if (matcher.match(data.key).size > 0) {
       matched.push(data.value);
     }
   });
@@ -176,9 +176,7 @@ LevelUpPersistence.prototype.storeSubscriptions = function(client, done) {
         qos: subscriptions[key].qos
       };
       var levelKey = util.format("%s:%s", key, client.id);
-      if (that._subMatcher.match(key).indexOf(levelKey) < 0) {
-        that._subMatcher.add(key, levelKey);
-      }
+      that._subMatcher.add(key, levelKey);
       that._subscriptions.put(levelKey, sub);
     });
   } else if (done) {
@@ -232,8 +230,7 @@ LevelUpPersistence.prototype.lookupSubscriptions = function(client, done) {
 LevelUpPersistence.prototype.storeOfflinePacket = function(packet, done) {
   var that = this;
   var subs = this._subMatcher.match(packet.topic);
-
-  async.each(subs, function(key, cb) {
+  async.each(Array.from(subs), function(key, cb) {
     that._subscriptions.get(key, function(err, sub) {
       if (err) {
         return cb(err);

diff --git a/lib/persistence/matcher.js b/lib/persistence/matcher.js
@@ -23,7 +23,7 @@ FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 OTHER DEALINGS IN THE SOFTWARE.
 */
 
-var Qlobber = require("qlobber").Qlobber;
+var Qlobber = require("qlobber").QlobberDedup;
 var util = require("util");
 
 function Matcher() {

diff --git a/lib/persistence/mongo.js b/lib/persistence/mongo.js
@@ -279,7 +279,7 @@ MongoPersistence.prototype.lookupRetained = function(pattern, cb) {
   });
 
   stream.on("data", function(data) {
-    if (matcher.match(data.topic).length > 0) {
+    if (matcher.match(data.topic).size > 0) {
       data.payload = data.payload.buffer;
       matched.push(data);
     }

diff --git a/lib/persistence/redis.js b/lib/persistence/redis.js
@@ -105,9 +105,7 @@ function RedisPersistence(options, callback) {
       }
 
       Object.keys(subs).forEach(function(sub) {
-        if (that._subMatcher.match(sub).indexOf(id) < 0) {
-          that._subMatcher.add(sub, id);
-        }
+        that._subMatcher.add(sub, id);
       });
 
       if( unsubs ) {
@@ -229,7 +227,7 @@ RedisPersistence.prototype.lookupRetained = function(pattern, done) {
     this._client.hkeys("retained", function(err, topics) {
       topics.sort();
       topics = topics.filter(function(topic) {
-        return matcher.match(topic).length > 0;
+        return matcher.match(topic).size > 0;
       });
 
       async.each(topics, match, function(err) {
@@ -286,9 +284,7 @@ RedisPersistence.prototype.storeSubscriptions = function(client, cb) {
     .pexpire(clientSubKey, this.options.ttl.subscriptions);
 
   Object.keys(subscriptions).forEach(function(e) {
-    if (that._subMatcher.match(e).indexOf(client.id) < 0) {
-      that._subMatcher.add(e, client.id);
-    }
+    that._subMatcher.add(e, client.id);
   });
 
   op.exec(cb);
@@ -353,7 +349,7 @@ RedisPersistence.prototype.storeOfflinePacket = function(packet, done) {
   var that = this;
 
   var matches = this._subMatcher.match(packet.topic);
-  async.each(matches, function(client, cb) {
+  async.each(Array.from(matches), function(client, cb) {
     that._storePacket(client, packet, cb);
   }, done);
 };