Class: Sidekiq::Throttled::Strategy

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/throttled/strategy.rb,
lib/sidekiq/throttled/strategy/base.rb,
lib/sidekiq/throttled/strategy/threshold.rb,
lib/sidekiq/throttled/strategy/concurrency.rb

Overview

Meta-strategy that couples Concurrency and Threshold strategies.

Defined Under Namespace

Modules: Base Classes: Concurrency, Threshold

Constant Summary collapse

VALID_VALUES_FOR_REQUEUE_WITH =

:enqueue means put the job back at the end of the queue immediately :schedule means schedule enqueueing the job for a later time when we expect to have capacity

i[enqueue schedule].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, concurrency: nil, threshold: nil, key_suffix: nil, observer: nil, requeue: nil) ⇒ Strategy

Returns a new instance of Strategy.

Parameters:

  • name (#to_s)
  • concurrency (Hash) (defaults to: nil) —

    Concurrency options. See keyword args of Sidekiq::Throttled::Strategy::Concurrency#initialize for details.

  • threshold (Hash) (defaults to: nil) —

    Threshold options. See keyword args of Sidekiq::Throttled::Strategy::Threshold#initialize for details.

  • key_suffix (#call) (defaults to: nil) —

    Dynamic key suffix generator.

  • observer (#call) (defaults to: nil) —

    Process called after throttled.

  • requeue (#call) (defaults to: nil) —

    What to do with jobs that are throttled.



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/sidekiq/throttled/strategy.rb', line 43

def initialize(name, concurrency: nil, threshold: nil, key_suffix: nil, observer: nil, requeue: nil) # rubocop:disable Metrics/MethodLength, Metrics/ParameterLists
  @observer = observer

  @concurrency = StrategyCollection.new(concurrency,
    strategy:   Concurrency,
    name:       name,
    key_suffix: key_suffix)

  @threshold = StrategyCollection.new(threshold,
    strategy:   Threshold,
    name:       name,
    key_suffix: key_suffix)

  @requeue_options = Throttled.config.default_requeue_options.merge(requeue || {})

  validate!
end

Instance Attribute Details

#concurrency ⇒ Object (readonly)

Returns the value of attribute concurrency.



21
22
23
 
# File 'lib/sidekiq/throttled/strategy.rb', line 21

def concurrency
  @concurrency
end

#observer ⇒ Object (readonly)

Returns the value of attribute observer.



29
30
31
 
# File 'lib/sidekiq/throttled/strategy.rb', line 29

def observer
  @observer
end

#requeue_options ⇒ Object (readonly)

Returns the value of attribute requeue_options.



33
34
35
 
# File 'lib/sidekiq/throttled/strategy.rb', line 33

def requeue_options
  @requeue_options
end

#threshold ⇒ Object (readonly)

Returns the value of attribute threshold.



25
26
27
 
# File 'lib/sidekiq/throttled/strategy.rb', line 25

def threshold
  @threshold
end

Instance Method Details

#dynamic? ⇒ Boolean

Returns whenever strategy has dynamic config.

Returns:

  • (Boolean) —

    whenever strategy has dynamic config



62
63
64
65
66
67
 
# File 'lib/sidekiq/throttled/strategy.rb', line 62

def dynamic?
  return true if @concurrency&.dynamic?
  return true if @threshold&.dynamic?

  false
end

#finalize!(jid, *job_args) ⇒ void

This method returns an undefined value.

Marks job as being processed.



117
118
119
 
# File 'lib/sidekiq/throttled/strategy.rb', line 117

def finalize!(jid, *job_args)
  @concurrency&.finalize!(jid, *job_args)
end

#requeue_throttled(work) ⇒ void

This method returns an undefined value.

Return throttled job to be executed later. Implementation depends on the strategy's requeue options.



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/sidekiq/throttled/strategy.rb', line 98

def requeue_throttled(work) # rubocop:disable Metrics/MethodLength
  # Resolve :with and :to options, calling them if they are Procs
  job_args = JSON.parse(work.job)["args"]
  with = requeue_with.respond_to?(:call) ? requeue_with.call(*job_args) : requeue_with
  target_queue = calc_target_queue(work)

  case with
  when :enqueue
    re_enqueue_throttled(work, target_queue)
  when :schedule
    # Find out when we will next be able to execute this job, and reschedule for then.
    reschedule_throttled(work, target_queue)
  else
    raise "unrecognized :with option #{with}"
  end
end

#requeue_to ⇒ String?

Returns Name of the queue to re-queue the job to.

Returns:

  • (String, nil) —

    Name of the queue to re-queue the job to.



92
93
94
 
# File 'lib/sidekiq/throttled/strategy.rb', line 92

def requeue_to
  requeue_options[:to]
end

#requeue_with ⇒ Proc, Symbol

Returns How to requeue the throttled job.

Returns:

  • (Proc, Symbol) —

    How to requeue the throttled job



87
88
89
 
# File 'lib/sidekiq/throttled/strategy.rb', line 87

def requeue_with
  requeue_options[:with]
end

#reset! ⇒ void

This method returns an undefined value.

Resets count of jobs of all available strategies



123
124
125
126
 
# File 'lib/sidekiq/throttled/strategy.rb', line 123

def reset!
  @concurrency&.reset!
  @threshold&.reset!
end

#throttled?(jid, *job_args) ⇒ Boolean

Returns whenever job is throttled or not.

Returns:

  • (Boolean) —

    whenever job is throttled or not.



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/sidekiq/throttled/strategy.rb', line 70

def throttled?(jid, *job_args)
  if @concurrency&.throttled?(jid, *job_args)
    @observer&.call(:concurrency, *job_args)
    return true
  end

  if @threshold&.throttled?(*job_args)
    @observer&.call(:threshold, *job_args)

    finalize!(jid, *job_args)
    return true
  end

  false
end