Class: Sidekiq::PriorityQueue::CombinedFetch

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/priority_queue/combined_fetch.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(fetches = []) ⇒ CombinedFetch

Returns a new instance of CombinedFetch.



8
9
10
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 8

def initialize(fetches = [])
  @fetches = fetches
end

Instance Attribute Details

#fetches ⇒ Object (readonly)

Returns the value of attribute fetches.



6
7
8
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 6

def fetches
  @fetches
end

Class Method Details

.configure {|combined_fetch| ... } ⇒ Object

Yields:

  • (combined_fetch) 


19
20
21
22
23
24
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 19

def self.configure(&block)
  combined_fetch = self.new
  yield combined_fetch

  combined_fetch
end

Instance Method Details

#add(fetch) ⇒ Object 



26
27
28
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 26

def add(fetch)
  fetches << fetch
end

#bulk_requeue(inprogress, options) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 30

def bulk_requeue(inprogress, options)
  # ReliableFetch#bulk_equeue ignores inprogress, so it's safe to call both
  fetches.each do |f|
    if [Fetch, ReliableFetch].any? { |klass| f.is_a?(klass) }
      jobs_to_requeue = inprogress.select{|uow| uow.queue.start_with?('priority-queue:') }
      f.bulk_requeue(jobs_to_requeue, options)
    else
      jobs_to_requeue = inprogress.reject{|uow| uow.queue.start_with?('priority-queue:') }
      f.bulk_requeue(jobs_to_requeue, options)
    end
  end
end

#retrieve_work ⇒ Object 



12
13
14
15
16
17
 
# File 'lib/sidekiq/priority_queue/combined_fetch.rb', line 12

def retrieve_work
  fetches.each do |fetch|
    work = fetch.retrieve_work
    return work if work
  end
end