Optimizes looking up all applications

Routinely, we load all applications and then load their individual data
hashes to get the subscriptions. This creates an N+1 number of requests
when dispatching requests. Pipelining should minimize network traffic.

Author: choubacha

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+- Pipelines fetching all queue subscriptions when using `QueueBus::Application.all`
+
 ## [0.11.0]
 
 ### Added

lib/queue_bus/application.rb

@@ -7,7 +7,22 @@ class Application
     class << self
       def all
         # note the names arent the same as we started with
-        ::QueueBus.redis { |redis| redis.smembers(app_list_key).collect { |val| new(val) } }
+        ::QueueBus.redis do |redis|
+          app_keys = redis.smembers(app_list_key)
+          apps = app_keys.collect { |val| new(val) }
+
+          hashes = redis.pipelined do
+            apps.each do |app|
+              redis.hgetall(app.redis_key)
+            end
+          end
+
+          apps.zip(hashes).each do |app, hash|
+            app._hydrate_redis_hash(hash)
+          end
+
+          apps
+        end
       end
     end
 
@@ -90,6 +105,10 @@ def event_display_tuples
       out
     end
 
+    def _hydrate_redis_hash(hash)
+      @raw_redis_hash = hash
+    end
+
     protected
 
     def self.normalize(val)
@@ -114,16 +133,24 @@ def subscriptions
 
     def read_redis_hash
       out = {}
-      ::QueueBus.redis do |redis|
-        redis.hgetall(redis_key).each do |key, val|
-          begin
-            out[key] = ::QueueBus::Util.decode(val)
-          rescue ::QueueBus::Util::DecodeException
-            out[key] = val
-          end
+      raw_redis_hash.each do |key, val|
+        begin
+          out[key] = ::QueueBus::Util.decode(val)
+        rescue ::QueueBus::Util::DecodeException
+          out[key] = val
         end
       end
       out
     end
+
+    private
+
+    def raw_redis_hash
+      return @raw_redis_hash if @raw_redis_hash
+
+      ::QueueBus.redis do |redis|
+        redis.hgetall(redis_key)
+      end
+    end
   end
 end