class Bunny::ConsumerWorkPool
Thread pool that dispatches consumer deliveries. Not supposed to be shared between channels or threads.
Every channel its own consumer pool.
@private
Attributes
API
Public Class Methods
Source
# File lib/bunny/consumer_work_pool.rb, line 22 def initialize(size = 1, abort_on_exception = false, shutdown_timeout = 60) @size = size @abort_on_exception = abort_on_exception @shutdown_timeout = shutdown_timeout @shutdown_mutex = ::Mutex.new @shutdown_conditional = ::ConditionVariable.new @queue = ::Queue.new @paused = false @running = false end
Public Instance Methods
Source
# File lib/bunny/consumer_work_pool.rb, line 79 def join(timeout = nil) (@threads || []).each { |t| t.join(timeout) } end
Source
# File lib/bunny/consumer_work_pool.rb, line 95 def kill @running = false (@threads || []).each { |t| t.kill } end
Source
# File lib/bunny/consumer_work_pool.rb, line 83 def pause @running = false @paused = true end
Source
# File lib/bunny/consumer_work_pool.rb, line 88 def resume @running = true @paused = false @threads.each { |t| t.run } end
Source
# File lib/bunny/consumer_work_pool.rb, line 62 def shutdown(wait_for_workers = false) was_running = running? @running = false @size.times do submit do |*args| throw :terminate end end return if !(wait_for_workers && @shutdown_timeout && was_running) @shutdown_mutex.synchronize do @shutdown_conditional.wait(@shutdown_mutex, @shutdown_timeout) if busy? end end
Source
# File lib/bunny/consumer_work_pool.rb, line 38 def start @threads = [] @size.times do t = Thread.new(&method(:run_loop)) t.abort_on_exception = true if abort_on_exception @threads << t end @running = true end
Source
# File lib/bunny/consumer_work_pool.rb, line 34 def submit(callable = nil, &block) @queue.push(callable || block) end
Protected Instance Methods
Source
# File lib/bunny/consumer_work_pool.rb, line 103 def run_loop catch(:terminate) do loop do Thread.stop if @paused callable = @queue.pop begin callable.call rescue ::StandardError => e # TODO: use connection logger $stderr.puts e.class.name $stderr.puts e.message end end end @shutdown_mutex.synchronize do @shutdown_conditional.signal unless busy? end end