Merge pull request #518 from adpdigital/fixLoadSubs

fix loadSubs callback issue & use pipeline to get subscriptions

lib/persistence/redis.js

@@ -98,6 +98,8 @@ function RedisPersistence(options, callback) {
       if (!result || typeof subs !== 'object') {
         if (!retried) {
           setTimeout(fetchAndUpdateLocalSub.bind(null, key, unsubs, true, cb), 500);
+        } else {
+          cb && cb();
         }
         return;
       }
@@ -132,24 +134,43 @@ function RedisPersistence(options, callback) {
       return;
     }
     var subsStream = that._client.scanStream({
-      match: "client:sub:*"
+      match: "client:sub:*",
+      count: 25000
     });
-    var keys = [];
+    var pipeline = that._client.pipeline();
+    var total = 0;
+    var done = null;
+
     subsStream.on('data', function(moreKeys){
-      for( var i=0; i<moreKeys.length; i++) {
-        keys.push(moreKeys[i]);
-      }
+      total += moreKeys.length;
+      moreKeys.map(function(k){
+        pipeline.get(k, function(err, result) {
+          if (err) {
+            done && done(err);
+            return;
+          }
+          var subs = JSON.parse(result);
+          if (!result || typeof subs !== 'object') {
+            done && done();
+            return;
+          }
+          updateLocalSub(k, subs);
+          done && done();
+        });
+      });
     });
+
     subsStream.on('end', function(){
-      steed.each(keys, function(k,next){
-        fetchAndUpdateLocalSub(k,null,false,next);
-      }, function(err) {
-        if (callback) {
-          callback(err, that);
+      if (total === 0) {
+        return callback(null, that);
+      }
+      done = function() {
+        if (--total === 0 && callback) {
+          callback(null, that);
           callback = null;
-          return;
         }
-      });
+      };
+      pipeline.exec();
     });
   });