Sha256: be4a2381d4aa850c23d7a7db4611cdde0509193129b56172cb0182e13f81c86f
Contents?: true
Size: 1.32 KB
Versions: 1
Compression:
Stored size: 1.32 KB
Contents
require 'concurrent/cached_thread_pool' 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) log(:info) { "Initialize new dispatcher (#{config[:max]} threads)..." } @pool = ::Concurrent::CachedThreadPool.new(max: config[:max]) 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 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 end end
Version data entries
1 entries across 1 versions & 1 rubygems
Version | Path |
---|---|
msgr-0.10.2 | lib/msgr/dispatcher.rb |