Class: Subscriber

Inherits:

Object

• Object
• Subscriber

Defined in:

lib/esp-commons/libraries/subscriber.rb

Instance Method Summary

• #logger ⇒ Object
• #start ⇒ Object
• #subscribers ⇒ Object

Instance Method Details

#logger ⇒ Object

# File 'lib/esp-commons/libraries/subscriber.rb', line 46

def logger
  @logger ||= ActiveSupport::BufferedLogger.new("#{Rails.root}/log/subscriber.log")
end

#start ⇒ Object

# File 'lib/esp-commons/libraries/subscriber.rb', line 4

def start
  AMQP.start(Settings['amqp.url']) do |connection|
    logger.debug "Subscribers runned in #{Rails.env} environment"
    subscribers.each do |subscriber|
      channel = AMQP::Channel.new(connection)
      channel.prefetch(1)

      topic = routing_prefix(subscriber)

      exchange = channel.topic(topic)
      queue = channel.queue(topic, :durable => true)

      queue.bind(exchange, :routing_key => "*").subscribe(:ack => true) do |header, message|
        method = header.routing_key
        message = JSON.parse(message)
        logger.debug "#{subscriber.class} receive #{method}: #{message}"
        if subscriber.respond_to?(method)
          begin
            subscriber.send(method, *message)
            logger.debug "#{subscriber.class} successfully executed #{method}"
          rescue => e
            logger.warn "#{subscriber.class} error while executing #{method} - #{e.message}"
          end
        else
          logger.warn "#{subscriber.class} cann't execute #{method} due #{subscriber.class} not respond to #{method}"
        end
        header.ack
      end
      logger.info "#{subscriber.class} listen #{queue.name}"
    end
  end
end

#subscribers ⇒ Object

# File 'lib/esp-commons/libraries/subscriber.rb', line 37

def subscribers
  Dir.glob("#{Rails.root}/lib/subscribers/*_subscriber.rb").map do |subscriber_path|
    require subscriber_path
    File.basename(subscriber_path, '.rb').classify.constantize.new.tap do |subscriber|
      subscriber.logger = logger if subscriber.respond_to? :logger=
    end
  end
end