Contents

module BBQueue
  class Consumer
    def initialize(queue_names, options = {})
      logger = options[:logger] || BBQueue::NullLogger.new

      Stalking::Consumer.new options.merge(:logger => BBQueue::FatalLogger.new(logger)) do
        Array(queue_names).each do |queue_name|
          job queue_name do |args|
            BBQueue::Serializer.load(args["object"]).work
          end
        end

        before do |queue_name, args|
          logger.info "Job #{BBQueue::Serializer.load(args["object"]).inspect} on #{queue_name.inspect} started"
        end

        after do |queue_name, args|
          logger.info "Job #{BBQueue::Serializer.load(args["object"]).inspect} on #{queue_name.inspect} finished"
        end

        error do |e, queue_name, args|
          logger.error "Job #{BBQueue::Serializer.load(args["object"]).inspect} on #{queue_name.inspect} failed"
          logger.error e
        end
      end
    end
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
bbqueue-0.0.1	lib/bbqueue/consumer.rb