Class: Sidekiq::ThrottledWorker::Fetch

Inherits:
BasicFetch
  • Object
show all
Defined in:
lib/sidekiq/throttled_worker/fetch.rb

Instance Method Summary collapse

Instance Method Details

#original_retrieve_work ⇒ Object 



4
 
# File 'lib/sidekiq/throttled_worker/fetch.rb', line 4

alias original_retrieve_work retrieve_work

#retrieve_work ⇒ Object 



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/sidekiq/throttled_worker/fetch.rb', line 6

def retrieve_work
  work = original_retrieve_work
  return work if work.nil?

  job_hash = Sidekiq.load_json(work.job)
  worker_class = Object.const_get(job_hash['class'])
  return work if worker_class.get_sidekiq_options["concurrency"].nil?

  if Concurrency.limit?(worker_class, job_hash['jid'])
    job_hash["concurrency_limit_count"] ||= 0
    job_hash["concurrency_limit_count"] += 1
    sleep(rand * [(job_hash["concurrency_limit_count"] - 1) * 0.01, 0.1].min)
    Sidekiq.redis do |conn|
      conn.lpush("queue:#{work.queue_name}", Sidekiq.dump_json(job_hash)) # put in queue end to auto wait some time in busy sidekiq cluster
    end
    nil
  else
    work
  end
end