Class: Sidekiq::Fetcher
- Inherits:
-
Object
- Object
- Sidekiq::Fetcher
- Includes:
- Celluloid, Util
- Defined in:
- lib/sidekiq/fetch.rb
Overview
The Fetcher blocks on Redis, waiting for a message to process from the queues. It gets the message and hands it to the Manager to assign to a ready Processor.
Constant Summary collapse
- TIMEOUT =
1
Constants included from Util
Class Method Summary collapse
Instance Method Summary collapse
-
#fetch ⇒ Object
Fetching is straightforward: the Manager makes a fetch request for each idle processor when Sidekiq starts and then issues a new fetch request every time a Processor finishes a message.
-
#initialize(mgr, options) ⇒ Fetcher
constructor
A new instance of Fetcher.
Methods included from Util
#hostname, #logger, #process_id, #redis, #watchdog
Methods included from ExceptionHandler
Constructor Details
Class Method Details
.done! ⇒ Object
Ugh. Say hello to a bloody hack. Can’t find a clean way to get the fetcher to just stop processing its mailbox when shutdown starts.
52 53 54 |
# File 'lib/sidekiq/fetch.rb', line 52 def self.done! @done = true end |
.done? ⇒ Boolean
56 57 58 |
# File 'lib/sidekiq/fetch.rb', line 56 def self.done? @done end |
.strategy ⇒ Object
60 61 62 |
# File 'lib/sidekiq/fetch.rb', line 60 def self.strategy Sidekiq.[:fetch] || BasicFetch end |
Instance Method Details
#fetch ⇒ Object
Fetching is straightforward: the Manager makes a fetch request for each idle processor when Sidekiq starts and then issues a new fetch request every time a Processor finishes a message.
Because we have to shut down cleanly, we can’t block forever and we can’t loop forever. Instead we reschedule a new fetch if the current fetch turned up nothing.
28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
# File 'lib/sidekiq/fetch.rb', line 28 def fetch watchdog('Fetcher#fetch died') do return if Sidekiq::Fetcher.done? begin work = @strategy.retrieve_work if work @mgr.async.assign(work) else after(0) { fetch } end rescue => ex logger.error("Error fetching message: #{ex}") logger.error(ex.backtrace.first) sleep(TIMEOUT) after(0) { fetch } end end end |