Sha256: 4571a6683bc327dcc35bfc01ab7a22c071300735e370bcbc3207bc502d4d0f5d
Contents?: true
Size: 1.43 KB
Versions: 13
Compression:
Stored size: 1.43 KB
Contents
require "thread"
module Bunny
# Thread pool that dispatches consumer deliveries. Not supposed to be shared between channels
# or threads.
#
# Every channel its own consumer pool.
#
# @private
class ConsumerWorkPool
#
# API
#
attr_reader :threads
attr_reader :size
def initialize(size = 1)
@size = size
@queue = ::Queue.new
end
def submit(callable = nil, &block)
@queue.push(callable || block)
end
def start
@threads = []
@size.times do
t = Thread.new(&method(:run_loop))
@threads << t
end
@running = true
end
def running?
@running
end
def shutdown
@running = false
@size.times do
submit do |*args|
throw :terminate
end
end
end
def join(timeout = nil)
@threads.each { |t| t.join(timeout) }
end
def pause
@running = false
@threads.each { |t| t.stop }
end
def resume
@running = true
@threads.each { |t| t.run }
end
def kill
@running = false
@threads.each { |t| t.kill }
end
protected
def run_loop
catch(:terminate) do
loop do
callable = @queue.pop
begin
callable.call
rescue Exception => e
# TODO
puts e.class.name
puts e.message
end
end
end
end
end
end
Version data entries
13 entries across 13 versions & 1 rubygems