Sidekiq redis-client gem compatibility

Author: mayesa

lib/sidekiq/batch.rb

@@ -16,15 +16,11 @@ class NoBlockGivenError < StandardError; end
 
     def initialize(existing_bid = nil)
       @bid = existing_bid || SecureRandom.urlsafe_base64(10)
-      @existing = !(!existing_bid || existing_bid.empty?)  # Basically existing_bid.present?
+      @existing = !(!existing_bid || existing_bid.empty?) # Basically existing_bid.present?
       @initialized = false
       @created_at = Time.now.utc.to_f
-      @bidkey = "BID-" + @bid.to_s
-      @queued_jids = []
-      @pending_jids = []
-      sidekiq_config = Sidekiq.const_defined?('MAJOR') && Sidekiq::MAJOR >= 7 ? (Thread.current[:sidekiq_capsule]&.config || Sidekiq.default_configuration) : Sidekiq.options
-      @incremental_push = sidekiq_config[:batch_push_interval]&.present?
-      @batch_push_interval = sidekiq_config[:batch_push_interval]
+      @bidkey = 'BID-' + @bid.to_s
+      @ready_to_queue = []
     end
 
     def description=(description)
@@ -43,14 +39,13 @@ def callback_batch=(callback_batch)
     end
 
     def on(event, callback, options = {})
-      return unless %w(success complete).include?(event.to_s)
+      return unless %w[success complete].include?(event.to_s)
+
       callback_key = "#{@bidkey}-callbacks-#{event}"
       Sidekiq.redis do |r|
         r.multi do |pipeline|
-          pipeline.sadd(callback_key, [JSON.unparse({
-            callback: callback,
-            opts: options
-          })])
+          json_params = JSON.unparse({ callback:, opts: options })
+          pipeline.sadd(callback_key, json_params)
           pipeline.expire(callback_key, BID_EXPIRE_TTL)
         end
       end
@@ -59,95 +54,64 @@ def on(event, callback, options = {})
     def jobs
       raise NoBlockGivenError unless block_given?
 
-      bid_data, Thread.current[:bid_data] = Thread.current[:bid_data], []
+      bid_data = Thread.current[:bid_data]
+      Thread.current[:bid_data] = []
 
       begin
         if !@existing && !@initialized
           parent_bid = Thread.current[:batch].bid if Thread.current[:batch]
 
           Sidekiq.redis do |r|
             r.multi do |pipeline|
-              pipeline.hset(@bidkey, "created_at", @created_at)
+              pipeline.hset(@bidkey, 'created_at', @created_at)
+              pipeline.hset(@bidkey, 'parent_bid', parent_bid.to_s) if parent_bid
               pipeline.expire(@bidkey, BID_EXPIRE_TTL)
-              if parent_bid
-                pipeline.hset(@bidkey, "parent_bid", parent_bid.to_s)
-                pipeline.hincrby("BID-#{parent_bid}", "children", 1)
-              end
             end
           end
 
           @initialized = true
         end
 
-        @queued_jids = []
-        @pending_jids = []
+        @ready_to_queue = []
 
         begin
           parent = Thread.current[:batch]
           Thread.current[:batch] = self
-          Thread.current[:parent_bid] = parent_bid
           yield
         ensure
           Thread.current[:batch] = parent
-          Thread.current[:parent_bid] = nil
         end
 
-        return [] if @queued_jids.size == 0
-        conditional_redis_increment!(true)
+        return [] if @ready_to_queue.size == 0
 
         Sidekiq.redis do |r|
           r.multi do |pipeline|
             if parent_bid
+              pipeline.hincrby("BID-#{parent_bid}", 'children', 1)
+              pipeline.hincrby("BID-#{parent_bid}", 'total', @ready_to_queue.size)
               pipeline.expire("BID-#{parent_bid}", BID_EXPIRE_TTL)
             end
 
+            pipeline.hincrby(@bidkey, 'pending', @ready_to_queue.size)
+            pipeline.hincrby(@bidkey, 'total', @ready_to_queue.size)
             pipeline.expire(@bidkey, BID_EXPIRE_TTL)
 
-            pipeline.sadd(@bidkey + "-jids", [@queued_jids.flatten])
-            pipeline.expire(@bidkey + "-jids", BID_EXPIRE_TTL)
+            # binding.break
+            @ready_to_queue.each do |ready_to_queue|
+              pipeline.sadd(@bidkey + '-jids', ready_to_queue)
+            end
+            pipeline.expire(@bidkey + '-jids', BID_EXPIRE_TTL)
           end
         end
 
-        @queued_jids
+        @ready_to_queue
       ensure
         Thread.current[:bid_data] = bid_data
       end
     end
 
     def increment_job_queue(jid)
-      @queued_jids << jid
-      @pending_jids << jid
-      conditional_redis_increment!
-    end
-
-    def conditional_redis_increment!(force=false)
-      if should_increment? || force
-        parent_bid = Thread.current[:parent_bid]
-        Sidekiq.redis do |r|
-          r.multi do |pipeline|
-            if parent_bid
-              pipeline.hincrby("BID-#{parent_bid}", "total", @pending_jids.length)
-              pipeline.expire("BID-#{parent_bid}", BID_EXPIRE_TTL)
-            end
-
-            pipeline.hincrby(@bidkey, "pending", @pending_jids.length)
-            pipeline.hincrby(@bidkey, "total", @pending_jids.length)
-            pipeline.expire(@bidkey, BID_EXPIRE_TTL)
-          end
-        end
-        @pending_jids = []
-      end
-    end
-
-    def should_increment?
-      return false unless @incremental_push
-      return true if @batch_push_interval == 0 || @queued_jids.length == 1
-      now = Time.now.to_f
-      @last_increment ||= now
-      if @last_increment + @batch_push_interval > now
-        @last_increment = now
-        return true
-      end
+      @ready_to_queue << jid
     end
 
     def invalidate_all
@@ -158,14 +122,14 @@ def invalidate_all
 
     def parent_bid
       Sidekiq.redis do |r|
-        r.hget(@bidkey, "parent_bid")
+        r.hget(@bidkey, 'parent_bid')
       end
     end
 
     def parent
-      if parent_bid
-        Sidekiq::Batch.new(parent_bid)
-      end
+      return unless parent_bid
+
+      Sidekiq::Batch.new(parent_bid)
     end
 
     def valid?(batch = self)
@@ -178,6 +142,7 @@ def valid?(batch = self)
     def persist_bid_attr(attribute, value)
       Sidekiq.redis do |r|
         r.multi do |pipeline|
+          value = value.to_s if value.in? [true, false]
           pipeline.hset(@bidkey, attribute, value)
           pipeline.expire(@bidkey, BID_EXPIRE_TTL)
         end
@@ -188,13 +153,14 @@ class << self
       def process_failed_job(bid, jid)
         _, pending, failed, children, complete, parent_bid = Sidekiq.redis do |r|
           r.multi do |pipeline|
-            pipeline.sadd("BID-#{bid}-failed", [jid])
+            jids = [jid]
+            jids.each { |jid| pipeline.sadd("BID-#{bid}-failed", jid) }
 
-            pipeline.hincrby("BID-#{bid}", "pending", 0)
+            pipeline.hincrby("BID-#{bid}", 'pending', 0)
             pipeline.scard("BID-#{bid}-failed")
-            pipeline.hincrby("BID-#{bid}", "children", 0)
+            pipeline.hincrby("BID-#{bid}", 'children', 0)
             pipeline.scard("BID-#{bid}-complete")
-            pipeline.hget("BID-#{bid}", "parent_bid")
+            pipeline.hget("BID-#{bid}", 'parent_bid')
 
             pipeline.expire("BID-#{bid}-failed", BID_EXPIRE_TTL)
           end
@@ -204,41 +170,44 @@ def process_failed_job(bid, jid)
         if parent_bid
           Sidekiq.redis do |r|
             r.multi do |pipeline|
-              pipeline.hincrby("BID-#{parent_bid}", "pending", 1)
-              pipeline.sadd("BID-#{parent_bid}-failed", [jid])
+              pipeline.hincrby("BID-#{parent_bid}", 'pending', 1)
+              jids = [jid]
+              jids.each { |jid| pipeline.sadd("BID-#{parent_bid}-failed", jid) }                
               pipeline.expire("BID-#{parent_bid}-failed", BID_EXPIRE_TTL)
             end
           end
         end
 
-        if pending.to_i == failed.to_i && children == complete
-          enqueue_callbacks(:complete, bid)
-        end
+        return unless pending.to_i == failed.to_i && children == complete
+
+        enqueue_callbacks(:complete, bid)
       end
 
       def process_successful_job(bid, jid)
         failed, pending, children, complete, success, total, parent_bid = Sidekiq.redis do |r|
           r.multi do |pipeline|
             pipeline.scard("BID-#{bid}-failed")
-            pipeline.hincrby("BID-#{bid}", "pending", -1)
-            pipeline.hincrby("BID-#{bid}", "children", 0)
+            pipeline.hincrby("BID-#{bid}", 'pending', -1)
+            pipeline.hincrby("BID-#{bid}", 'children', 0)
             pipeline.scard("BID-#{bid}-complete")
             pipeline.scard("BID-#{bid}-success")
-            pipeline.hget("BID-#{bid}", "total")
-            pipeline.hget("BID-#{bid}", "parent_bid")
+            pipeline.hget("BID-#{bid}", 'total')
+            pipeline.hget("BID-#{bid}", 'parent_bid')
+
+            jids = [jid]
+            jids.each { |jid| pipeline.srem("BID-#{bid}-failed", jid) }
+            jids.each { |jid| pipeline.srem("BID-#{bid}-jids", jid) }
 
-            pipeline.srem("BID-#{bid}-failed", [jid])
-            pipeline.srem("BID-#{bid}-jids", [jid])
             pipeline.expire("BID-#{bid}", BID_EXPIRE_TTL)
           end
         end
 
         all_success = pending.to_i.zero? && children == success
         # if complete or successfull call complete callback (the complete callback may then call successful)
-        if (pending.to_i == failed.to_i && children == complete) || all_success
-          enqueue_callbacks(:complete, bid)
-          enqueue_callbacks(:success, bid) if all_success
-        end
+        return unless (pending.to_i == failed.to_i && children == complete) || all_success
+
+        enqueue_callbacks(:complete, bid)
+        enqueue_callbacks(:success, bid) if all_success
       end
 
       def enqueue_callbacks(event, bid)
@@ -250,30 +219,30 @@ def enqueue_callbacks(event, bid)
             pipeline.hget(batch_key, event_name)
             pipeline.hset(batch_key, event_name, 'true')
             pipeline.smembers(callback_key)
-            pipeline.hget(batch_key, "callback_queue")
-            pipeline.hget(batch_key, "parent_bid")
-            pipeline.hget(batch_key, "callback_batch")
+            pipeline.hget(batch_key, 'callback_queue')
+            pipeline.hget(batch_key, 'parent_bid')
+            pipeline.hget(batch_key, 'callback_batch')
           end
         end
 
         return if already_processed == 'true'
 
-        queue ||= "default"
-        parent_bid = !parent_bid || parent_bid.empty? ? nil : parent_bid    # Basically parent_bid.blank?
+        queue ||= 'default'
+        parent_bid = nil if !parent_bid || parent_bid.empty? # Basically parent_bid.blank?
         callback_args = callbacks.reduce([]) do |memo, jcb|
           cb = Sidekiq.load_json(jcb)
           memo << [cb['callback'], event_name, cb['opts'], bid, parent_bid]
         end
 
-        opts = {"bid" => bid, "event" => event_name}
+        opts = { 'bid' => bid, 'event' => event_name }
 
         # Run callback batch finalize synchronously
         if callback_batch
           # Extract opts from cb_args or use current
           # Pass in stored event as callback finalize is processed on complete event
           cb_opts = callback_args.first&.at(2) || opts
 
-          Sidekiq.logger.debug {"Run callback batch bid: #{bid} event: #{event_name} args: #{callback_args.inspect}"}
+          Sidekiq.logger.debug { "Run callback batch bid: #{bid} event: #{event_name} args: #{callback_args.inspect}" }
           # Finalize now
           finalizer = Sidekiq::Batch::Callback::Finalize.new
           status = Status.new bid
@@ -282,7 +251,7 @@ def enqueue_callbacks(event, bid)
           return
         end
 
-        Sidekiq.logger.debug {"Enqueue callback bid: #{bid} event: #{event_name} args: #{callback_args.inspect}"}
+        Sidekiq.logger.debug { "Enqueue callback bid: #{bid} event: #{event_name} args: #{callback_args.inspect}" }
 
         if callback_args.empty?
           # Finalize now
@@ -291,40 +260,41 @@ def enqueue_callbacks(event, bid)
           finalizer.dispatch(status, opts)
         else
           # Otherwise finalize in sub batch complete callback
-          cb_batch = self.new
+          cb_batch = new
           cb_batch.callback_batch = 'true'
-          Sidekiq.logger.debug {"Adding callback batch: #{cb_batch.bid} for batch: #{bid}"}
-          cb_batch.on(:complete, "Sidekiq::Batch::Callback::Finalize#dispatch", opts)
+          Sidekiq.logger.debug { "Adding callback batch: #{cb_batch.bid} for batch: #{bid}" }
+          cb_batch.on(:complete, 'Sidekiq::Batch::Callback::Finalize#dispatch', opts)
           cb_batch.jobs do
             push_callbacks callback_args, queue
           end
         end
       end
 
       def cleanup_redis(bid)
-        Sidekiq.logger.debug {"Cleaning redis of batch #{bid}"}
+        Sidekiq.logger.debug { "Cleaning redis of batch #{bid}" }
         Sidekiq.redis do |r|
           r.del(
             "BID-#{bid}",
             "BID-#{bid}-callbacks-complete",
             "BID-#{bid}-callbacks-success",
             "BID-#{bid}-failed",
-
             "BID-#{bid}-success",
             "BID-#{bid}-complete",
-            "BID-#{bid}-jids",
+            "BID-#{bid}-jids"
           )
         end
       end
 
-    private
+      private
+
+      def push_callbacks(args, queue)
+        return if args.empty?
 
-      def push_callbacks args, queue
         Sidekiq::Client.push_bulk(
           'class' => Sidekiq::Batch::Callback::Worker,
           'args' => args,
           'queue' => queue
-        ) unless args.empty?
+        )
       end
     end
   end

lib/sidekiq/batch/version.rb

@@ -1,5 +1,5 @@
 module Sidekiq
   class Batch
-    VERSION = '0.1.9'.freeze
+    VERSION = '0.2.0'.freeze
   end
 end