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