Merge pull request gzigzigzeo#54 from mrexox/feature/support-sidekiq-upto-7

Support sidekiq upto 7

Author: gzigzigzeo

.gitignore

@@ -16,3 +16,4 @@ spec/reports
 test/tmp
 test/version_tmp
 tmp
+.lefthook-local.yml

Appraisals

@@ -21,3 +21,11 @@ end
 appraise 'sidekiq-6.5' do
   gem 'sidekiq', '~> 6.5.0'
 end
+
+appraise 'sidekiq-7.0' do
+  gem 'sidekiq', '~> 7.0.0'
+end
+
+appraise 'sidekiq-master' do
+  gem 'sidekiq', github: 'mperham/sidekiq'
+end

bin/console

@@ -0,0 +1,8 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require "bundler/setup"
+require "sidekiq/grouping"
+
+require "pry"
+Pry.start

gemfiles/sidekiq_7.0.gemfile

@@ -0,0 +1,7 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "sidekiq", "~> 7.0.0"
+
+gemspec path: "../"

lefthook.yml

@@ -4,12 +4,17 @@
 # See: github.com/evilmartians/lefthook
 
 pre-commit:
+  parallel: true
   commands:
+    appraisal:
+      glob: "{Appraisals,*.gemfile}"
+      run: echo {staged_files} > /dev/null; bundle exec appraisal install && git add gemfiles/*.gemfile
     rubocop:
-      glob: "*.rb"
+      glob: "{*.rb,*.gemspec,Gemfile,Rakefile}"
       run: bundle exec rubocop -A {staged_files} && git add {staged_files}
 
 pre-push:
   commands:
     rspec:
-      run: bundle exec appraisal rspec
+      glob: "*.rb"
+      run: echo {push_files} > /dev/null; bundle exec appraisal rspec

lib/sidekiq/grouping.rb

@@ -4,6 +4,7 @@
 require "active_support/core_ext/string"
 require "active_support/configurable"
 require "active_support/core_ext/numeric/time"
+require "sidekiq"
 require "sidekiq/grouping/version"
 require "concurrent"
 

lib/sidekiq/grouping/config.rb

@@ -6,10 +6,12 @@ module Config
       include ActiveSupport::Configurable
 
       def self.options
-        if Sidekiq.respond_to?(:[])
-          Sidekiq[:grouping] || Sidekiq["grouping"] || {}
-        else
+        if Sidekiq.respond_to?(:[]) # Sidekiq 6.x
+          Sidekiq[:grouping] || {}
+        elsif Sidekiq.respond_to?(:options) # Sidekiq <= 5.x
           Sidekiq.options[:grouping] || Sidekiq.options["grouping"] || {}
+        else # Sidekiq 7.x
+          Sidekiq.default_configuration[:grouping] || {}
         end
       end
 

lib/sidekiq/grouping/redis.rb

@@ -1,8 +1,12 @@
 # frozen_string_literal: true
 
+require_relative "./redis_dispatcher"
+
 module Sidekiq
   module Grouping
     class Redis
+      include RedisDispatcher
+
       PLUCK_SCRIPT = <<-SCRIPT
         local pluck_values = redis.call('lpop', KEYS[1], ARGV[1]) or {}
         if #pluck_values > 0 then
@@ -15,61 +19,75 @@ def push_msg(name, msg, remember_unique: false)
         redis do |conn|
           conn.multi do |pipeline|
             sadd = pipeline.respond_to?(:sadd?) ? :sadd? : :sadd
-            pipeline.public_send(sadd, ns("batches"), name)
-            pipeline.rpush(ns(name), msg)
+            redis_connection_call(pipeline, sadd, ns("batches"), name)
+            redis_connection_call(pipeline, :rpush, ns(name), msg)
+
             if remember_unique
-              pipeline.public_send(
-                sadd,
-                unique_messages_key(name),
-                msg
+              redis_connection_call(
+                pipeline, sadd, unique_messages_key(name), msg
               )
             end
           end
         end
       end
 
       def enqueued?(name, msg)
-        redis do |conn|
-          conn.sismember(unique_messages_key(name), msg)
-        end
+        member = redis_call(:sismember, unique_messages_key(name), msg)
+        return member if member.is_a?(TrueClass) || member.is_a?(FalseClass)
+
+        member != 0
       end
 
       def batch_size(name)
-        redis { |conn| conn.llen(ns(name)) }
+        redis_call(:llen, ns(name))
       end
 
       def batches
-        redis { |conn| conn.smembers(ns("batches")) }
+        redis_call(:smembers, ns("batches"))
       end
 
       def pluck(name, limit)
-        keys = [ns(name), unique_messages_key(name)]
-        args = [limit]
-        redis { |conn| conn.eval PLUCK_SCRIPT, keys, args }
+        if new_redis_client?
+          redis_call(
+            :eval,
+            PLUCK_SCRIPT,
+            2,
+            ns(name),
+            unique_messages_key(name),
+            limit
+          )
+        else
+          keys = [ns(name), unique_messages_key(name)]
+          args = [limit]
+          redis_call(:eval, PLUCK_SCRIPT, keys, args)
+        end
       end
 
       def get_last_execution_time(name)
-        redis { |conn| conn.get(ns("last_execution_time:#{name}")) }
+        redis_call(:get, ns("last_execution_time:#{name}"))
       end
 
       def set_last_execution_time(name, time)
-        redis do |conn|
-          conn.set(ns("last_execution_time:#{name}"), time.to_json)
-        end
+        redis_call(
+          :set, ns("last_execution_time:#{name}"), time.to_json
+        )
       end
 
       def lock(name)
-        redis do |conn|
-          id = ns("lock:#{name}")
-          conn.set(id, true, nx: true, ex: Sidekiq::Grouping::Config.lock_ttl)
-        end
+        redis_call(
+          :set,
+          ns("lock:#{name}"),
+          "true",
+          nx: true,
+          ex: Sidekiq::Grouping::Config.lock_ttl
+        )
       end
 
       def delete(name)
         redis do |conn|
-          conn.del(ns("last_execution_time:#{name}"))
-          conn.del(ns(name))
-          conn.srem(ns("batches"), name)
+          redis_connection_call(conn, :del, ns("last_execution_time:#{name}"))
+          redis_connection_call(conn, :del, ns(name))
+          redis_connection_call(conn, :srem, ns("batches"), name)
         end
       end
 
@@ -82,10 +100,6 @@ def unique_messages_key(name)
       def ns(key = nil)
         "batching:#{key}"
       end
-
-      def redis(&block)
-        Sidekiq.redis(&block)
-      end
     end
   end
 end

lib/sidekiq/grouping/redis_dispatcher.rb

@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+module Sidekiq
+  module Grouping
+    module RedisDispatcher
+      def redis_call(command, *args, **kwargs)
+        redis do |connection|
+          redis_connection_call(connection, command, *args, **kwargs)
+        end
+      end
+
+      def redis_connection_call(connection, command, *args, **kwargs)
+        if new_redis_client? # redis-client
+          connection.call(command.to_s.upcase, *args, **kwargs)
+        else # redis
+          connection.public_send(command, *args, **kwargs)
+        end
+      end
+
+      def new_redis_client?
+        Sidekiq::VERSION[0].to_i >= 7
+      end
+
+      def redis(&block)
+        Sidekiq.redis(&block)
+      end
+    end
+  end
+end

lib/sidekiq/grouping/version.rb

@@ -2,6 +2,6 @@
 
 module Sidekiq
   module Grouping
-    VERSION = "1.2.0"
+    VERSION = "1.3.0"
   end
 end