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, queues, strict) ⇒ Fetcher
constructor
A new instance of Fetcher.
Methods included from Util
#hostname, #logger, #process_id, #redis, #watchdog
Methods included from ExceptionHandler
Constructor Details
#initialize(mgr, queues, strict) ⇒ Fetcher
Returns a new instance of Fetcher.
15 16 17 18 19 20 |
# File 'lib/sidekiq/fetch.rb', line 15 def initialize(mgr, queues, strict) @mgr = mgr @strictly_ordered_queues = strict @queues = queues.map { |q| "queue:#{q}" } @unique_queues = @queues.uniq end |
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.
54 55 56 |
# File 'lib/sidekiq/fetch.rb', line 54 def self.done! @done = true end |
.done? ⇒ Boolean
58 59 60 |
# File 'lib/sidekiq/fetch.rb', line 58 def self.done? @done 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.
30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 |
# File 'lib/sidekiq/fetch.rb', line 30 def fetch watchdog('Fetcher#fetch died') do return if Sidekiq::Fetcher.done? begin queue, msg = Sidekiq.redis { |conn| conn.blpop(*queues_cmd) } if msg @mgr.async.assign(msg, queue.gsub(/.*queue:/, '')) 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 |