Class: Msgr::Dispatcher
- Inherits:
-
Object
- Object
- Msgr::Dispatcher
- Includes:
- Logging
- Defined in:
- lib/msgr/dispatcher.rb
Overview
The Dispatcher receives incoming messages, process them through a middleware stack and delegate them to a new and fresh consumer instance.
Defined Under Namespace
Classes: NullPool
Instance Attribute Summary collapse
-
#config ⇒ Object
readonly
Returns the value of attribute config.
-
#pool ⇒ Object
readonly
Returns the value of attribute pool.
Instance Method Summary collapse
- #call(message) ⇒ Object
-
#dispatch(message) ⇒ Object
rubocop:disable Metrics/AbcSize rubocop:disable Metrics/MethodLength rubocop:disable Metrics/CyclomaticComplexity.
-
#initialize(config) ⇒ Dispatcher
constructor
A new instance of Dispatcher.
- #shutdown ⇒ Object
- #to_s ⇒ Object
Methods included from Logging
Constructor Details
#initialize(config) ⇒ Dispatcher
Returns a new instance of Dispatcher.
13 14 15 16 17 18 19 20 21 22 |
# File 'lib/msgr/dispatcher.rb', line 13 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 |
Instance Attribute Details
#config ⇒ Object (readonly)
Returns the value of attribute config.
11 12 13 |
# File 'lib/msgr/dispatcher.rb', line 11 def config @config end |
#pool ⇒ Object (readonly)
Returns the value of attribute pool.
11 12 13 |
# File 'lib/msgr/dispatcher.rb', line 11 def pool @pool end |
Instance Method Details
#call(message) ⇒ Object
24 25 26 27 28 |
# File 'lib/msgr/dispatcher.rb', line 24 def call() pool.post() do |msg| dispatch msg end end |
#dispatch(message) ⇒ Object
rubocop:disable Metrics/AbcSize rubocop:disable Metrics/MethodLength rubocop:disable Metrics/CyclomaticComplexity
33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
# File 'lib/msgr/dispatcher.rb', line 33 def dispatch() consumer_class = Object.const_get .route.consumer log(:debug) { "Dispatch message to #{consumer_class.name}" } consumer_class.new.dispatch # Acknowledge message unless it is already acknowledged or auto_ack is disabled. .ack unless .acked? or not consumer_class.auto_ack? rescue => error .nack unless .acked? log(:error) do "Dispatcher error: #{error.class.name}: #{error}\n" + error.backtrace.join("\n") end raise error 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 |
#shutdown ⇒ Object
59 |
# File 'lib/msgr/dispatcher.rb', line 59 def shutdown; end |
#to_s ⇒ Object
61 62 63 |
# File 'lib/msgr/dispatcher.rb', line 61 def to_s self.class.name end |