concurrency.rb in sidekiq-throttled-0.8.0

- old
+ new

@@ -26,11 +26,11 @@
         # @param [#to_i, #call] limit Amount of allowed concurrent jobs
         #   per processors running for given key.
         # @param [#to_i] ttl Concurrency lock TTL in seconds.
         # @param [Proc] key_suffix Dynamic key suffix generator.
         def initialize(strategy_key, limit:, ttl: 900, key_suffix: nil)
-          @base_key   = "#{strategy_key}:concurrency"
+          @base_key   = "#{strategy_key}:concurrency.v2"
           @limit      = limit
           @ttl        = ttl.to_i
           @key_suffix = key_suffix
         end
 
@@ -42,18 +42,18 @@
         # @return [Boolean] whenever job is throttled or not
         def throttled?(jid, *job_args)
           return false unless (job_limit = limit(job_args))
 
           keys = [key(job_args)]
-          args = [jid.to_s, job_limit, @ttl]
+          args = [jid.to_s, job_limit, @ttl, Time.now.to_f]
 
           1 == SCRIPT.eval(keys, args)
         end
 
         # @return [Integer] Current count of jobs
         def count(*job_args)
-          Sidekiq.redis { |conn| conn.scard(key(job_args)) }.to_i
+          Sidekiq.redis { |conn| conn.zcard(key(job_args)) }.to_i
         end
 
         # Resets count of jobs
         # @return [void]
         def reset!(*job_args)
@@ -61,10 +61,10 @@
         end
 
         # Remove jid from the pool of jobs in progress
         # @return [void]
         def finalize!(jid, *job_args)
-          Sidekiq.redis { |conn| conn.srem(key(job_args), jid.to_s) }
+          Sidekiq.redis { |conn| conn.zrem(key(job_args), jid.to_s) }
         end
       end
     end
   end
 end