Sha256: 32b2ffb32fd0e457f68acb79324b2fac95e1417744f84439d7a42bab615c1d98
Contents?: true
Size: 1.68 KB
Versions: 7
Compression:
Stored size: 1.68 KB
Contents
# frozen_string_literal: true 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 :config, :pool def initialize(config) config[:pool_class] ||= 'Msgr::Dispatcher::NullPool' log(:debug) do "Initialize new dispatcher (#{config[:pool_class]}: #{config})..." end @config = config @pool = config[:pool_class].constantize.new config end def call(message) pool.post(message) do |msg| dispatch msg 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 only if it is not already acknowledged and auto # acknowledgment is enabled. message.ack unless message.acked? || !consumer_class.auto_ack? rescue StandardError => e message.nack unless message.acked? log(:error) do "Dispatcher error: #{e.class.name}: #{e}\n" + e.backtrace.join("\n") end raise e if config[:raise_exceptions] 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