# frozen_string_literal: true # internal require "sidekiq/throttled/strategy/script" module Sidekiq module Throttled class Strategy # Threshold throttling strategy # @todo Use redis TIME command instead of sending current timestamp from # sidekiq manager. See: http://redis.io/commands/time class Threshold # LUA script used to limit fetch threshold. # Logic behind the scene can be described in following pseudo code: # # def exceeded? # @limit <= LLEN(@key) && NOW - LINDEX(@key, -1) < @period # end # # def increase! # LPUSH(@key, NOW) # LTRIM(@key, 0, @limit - 1) # EXPIRE(@key, @period) # end # # return 1 if exceeded? # # increase! # return 0 SCRIPT = Script.new File.read "#{__dir__}/threshold.lua" private_constant :SCRIPT # @!attribute [r] limit # @return [Integer] Amount of jobs allowed per period attr_reader :limit # @!attribute [r] period # @return [Float] Period in seconds attr_reader :period # @param [#to_s] strategy_key # @param [Hash] opts # @option opts [#to_i] :limit Amount of jobs allowed per period # @option opts [#to_f] :period Period in seconds def initialize(strategy_key, opts) @base_key = "#{strategy_key}:threshold".freeze @limit = opts.fetch(:limit).to_i @period = opts.fetch(:period).to_f @key_suffix = opts[:key_suffix] end def dynamic_keys? @key_suffix end # @return [Boolean] whenever job is throttled or not def throttled?(*job_args) 1 == SCRIPT.eval([key(job_args)], [@limit, @period, Time.now.to_f]) end # @return [Integer] Current count of jobs def count(*job_args) Sidekiq.redis { |conn| conn.llen(key(job_args)) }.to_i end # Resets count of jobs # @return [void] def reset!(*job_args) Sidekiq.redis { |conn| conn.del(key(job_args)) } end private def key(job_args) key = @base_key.dup key << ":#{@key_suffix.call(*job_args)}" if @key_suffix key end end end end end