Class: Jiggler::AtLeastOnce::Acknowledger

Inherits:
BaseAcknowledger show all
Defined in:
lib/jiggler/at_least_once/acknowledger.rb

Instance Method Summary collapse

Methods included from Support::Helper

#log_error, #log_error_short, #logger, #safe_async, #scan_all, #tid

Constructor Details

#initialize(config) ⇒ Acknowledger

Returns a new instance of Acknowledger.



6
7
8
9
10
 
# File 'lib/jiggler/at_least_once/acknowledger.rb', line 6

def initialize(config)
  super
  @runners = []
  @queue = Queue.new
end

Instance Method Details

#ack(job) ⇒ Object 



12
13
14
 
# File 'lib/jiggler/at_least_once/acknowledger.rb', line 12

def ack(job)
  @queue.push(job)
end

#start ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/jiggler/at_least_once/acknowledger.rb', line 16

def start
  @config[:concurrency].times do
    @runners << safe_async('Acknowledger') do
      while (job = @queue.pop) != nil
        begin
          job.ack
        rescue StandardError => err
          log_error(err, context: '\'Could not acknowledge a job\'', job: job)
        end
      end
      logger.debug('Acknowledger exits')
    rescue Async::Stop
      logger.debug('Acknowledger received stop signal')
    end
  end
end

#terminate ⇒ Object 



37
38
39
40
 
# File 'lib/jiggler/at_least_once/acknowledger.rb', line 37

def terminate
  logger.debug('Suspending the acknowledger')
  @queue.close
end

#wait ⇒ Object 



33
34
35
 
# File 'lib/jiggler/at_least_once/acknowledger.rb', line 33

def wait
  @runners.each(&:wait)
end