Class: SidekiqSolidFetch::UnitOfWork

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq_solid_fetch/unit_of_work.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue, job, config, processing_queue) ⇒ UnitOfWork

Returns a new instance of UnitOfWork.



5
6
7
8
9
10
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 5

def initialize(queue, job, config, processing_queue)
  @queue = queue
  @job = job
  @config = config
  @processing_queue = processing_queue
end

Instance Attribute Details

#config ⇒ Object

Returns the value of attribute config.



3
4
5
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 3

def config
  @config
end

#job ⇒ Object

Returns the value of attribute job.



3
4
5
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 3

def job
  @job
end

#processing_queue ⇒ Object

Returns the value of attribute processing_queue.



3
4
5
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 3

def processing_queue
  @processing_queue
end

#queue ⇒ Object

Returns the value of attribute queue.



3
4
5
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 3

def queue
  @queue
end

Instance Method Details

#acknowledge ⇒ Object 



12
13
14
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 12

def acknowledge
  config.redis { |conn| conn.lrem(processing_queue, -1, job) }
end

#queue_name ⇒ Object 



16
17
18
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 16

def queue_name
  queue.delete_prefix("queue:")
end

#requeue ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 20

def requeue
  config.redis do |conn|
    conn.multi do |multi|
      multi.lrem(processing_queue, -1, job)
      # RPUSH, like Sidekiq's own requeue: the queue is consumed from the
      # right, so an interrupted job runs next rather than going behind
      # newly enqueued work.
      multi.rpush(queue, job)
    end
  end
end

#requeue_throttled(target_queue) ⇒ Object

Atomically move the job from the processing queue to the BACK of the target queue (LPUSH side), so a throttled job doesn't churn straight through fetch again and no copy is left behind in the processing queue.



35
36
37
38
39
40
41
42
 
# File 'lib/sidekiq_solid_fetch/unit_of_work.rb', line 35

def requeue_throttled(target_queue)
  config.redis do |conn|
    conn.multi do |multi|
      multi.lrem(processing_queue, -1, job)
      multi.lpush(target_queue, job)
    end
  end
end