Sha256: d39ca5e0b091638d229745c3e5d5b34b8d856bd8c1142f22ce4d816d66e87da7
Contents?: true
Size: 1.49 KB
Versions: 7
Compression:
Stored size: 1.49 KB
Contents
module Msgr
# The Dispatcher receives incoming messages,
# process them through a middleware stack and
# delegate them to a new and fresh consumer instance.
#
class Dispatcher
include Logging
attr_reader :pool
def initialize(config)
config[:pool_class] ||= 'Msgr::Dispatcher::NullPool'
log(:debug) { "Initialize new dispatcher (#{config[:pool_class]}: #{config})..." }
@pool = config[:pool_class].constantize.new config
end
def call(message)
pool.post(message) do |message|
dispatch message
end
end
def dispatch(message)
consumer_class = Object.const_get message.route.consumer
log(:debug) { "Dispatch message to #{consumer_class.name}" }
consumer_class.new.dispatch message
# Acknowledge message unless it is already acknowledged.
message.ack unless message.acked?
rescue => error
message.nack unless message.acked?
log(:error) do
"Dispatcher error: #{error.class.name}: #{error}\n" +
error.backtrace.join("\n")
end
ensure
if defined?(ActiveRecord) && ActiveRecord::Base.connection_pool.active_connection?
log(:debug) { 'Release used AR connection for dispatcher thread.' }
ActiveRecord::Base.connection_pool.release_connection
end
end
def shutdown
end
def to_s
self.class.name
end
class NullPool
def initialize(*)
end
def post(*args)
yield(*args)
end
end
end
end
Version data entries
7 entries across 7 versions & 1 rubygems