Class: SidekiqSolidFetch::Fetcher

Inherits:
Object
  • Object
show all
Includes:
Sidekiq::Component
Defined in:
lib/sidekiq_solid_fetch/fetcher.rb

Constant Summary collapse

TIMEOUT =

How long to pause when every queue is empty. Mirrors BasicFetch's BRPOP timeout: Sidekiq's processor loop calls retrieve_work with no delay of its own, so blocking here is what keeps idle threads from hot-looping against Redis.

2

Instance Method Summary collapse

Constructor Details

#initialize(cap) ⇒ Fetcher

Returns a new instance of Fetcher.

Raises:

  • (ArgumentError) 


15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/sidekiq_solid_fetch/fetcher.rb', line 15

def initialize(cap)
  raise ArgumentError, "missing queue list" unless cap.queues
  @config = cap
  @strictly_ordered_queues = cap.mode == :strict
  @queues = config.queues.map { |q| "queue:#{q}" }
  @queues.uniq! if @strictly_ordered_queues
  @identity = cap.identity
  @processing_queues = @queues.uniq.to_h do |queue|
    [queue, ::SidekiqSolidFetch.processing_queue_name(queue, @identity, cap.name)]
  end
  register_processing_queues
end

Instance Method Details

#bulk_requeue ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/sidekiq_solid_fetch/fetcher.rb', line 42

def bulk_requeue(*)
  logger.info("SidekiqSolidFetch: Re-queueing terminated jobs for #{@identity}")

  count = 0
  redis do |conn|
    @processing_queues.each do |queue, processing_queue|
      while conn.lmove(processing_queue, queue, "LEFT", "RIGHT")
        count += 1
        logger.info { "SidekiqSolidFetch: Moving job from #{processing_queue} back to #{queue}" }
      end
    end
    ::SidekiqSolidFetch.unregister_processing_queues(conn, @processing_queues.values)
  end
  logger.info("SidekiqSolidFetch: Re-queued #{count} jobs")
rescue => ex
  logger.warn("SidekiqSolidFetch: Failed to requeue jobs: #{ex.message}")
end

#queues_cmd ⇒ Object 



60
61
62
63
64
65
66
67
68
 
# File 'lib/sidekiq_solid_fetch/fetcher.rb', line 60

def queues_cmd
  if @strictly_ordered_queues
    @queues
  else
    permute = @queues.shuffle
    permute.uniq!
    permute
  end
end

#retrieve_work ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/sidekiq_solid_fetch/fetcher.rb', line 28

def retrieve_work
  queues_cmd.each do |queue|
    processing_queue = @processing_queues[queue]
    work = redis do |conn|
      conn.lmove(queue, processing_queue, "RIGHT", "LEFT")
    end

    return ::SidekiqSolidFetch::UnitOfWork.new(queue, work, config, processing_queue) if work
  end

  sleep(TIMEOUT)
  nil
end