<commit_before>require 'amqp'
require 'multi_json'
require 'hashr'
require 'benchmark'

module Travis
  class Consumer
    autoload :Handler, 'travis/consumer/handler'
    autoload :Job,     'travis/consumer/job'
    autoload :Worker,  'travis/consumer/worker'

    include Logging

    REPORTING_KEY = 'reporting.jobs'

    class << self
      def start(options = {})
        Database.connect(options)

        EM.run do
          prune_workers!
          # cleanup_jobs
          subscribe
        end
      end

      def subscribe
        new.subscribe
      end

      def prune_workers!
        interval = Travis.config.workers.prune.interval
        EM.add_periodic_timer(interval, &::Worker.method(:prune))
      end

      def cleanup_jobs!
        interval = Travis.config.jobs.retry.interval
        EM.add_periodic_timer(interval, &::Job.method(:cleanup))
      end
    end

    attr_reader :config

    def initialize
      @config = Travis.config.amqp
    end

    def subscribe
      queue.subscribe(:ack => true, &method(:receive))
    end

    def receive(message, payload)
      log notice("Handling event #{message.type.inspect} with payload : #{payload.inspect}")

      event   = message.type
      payload = decode(payload)
      handler = handler_for(event, payload)

      benchmark_and_cache do
        handler.handle
      end

      message.ack
    rescue Exception => e
      puts e.message, e.backtrace
      message.ack
      # message.reject(:requeue => false) # how to decide whether to requeue the message?
    end

    def heartbeat(message, payload)
      log notice("Heartbeat: #{payload.inspect}")
      Worker.heartbeat
    end

    protected

      def handler_for(event, payload)
        case event.to_s
        when /^job/
          Job.new(event, payload)
        when /^worker/
          Worker.new(event, payload)
        else
          raise "Unknown message type: #{event.inspect}"
        end
      end

      def benchmark_and_cache
        timing = Benchmark.realtime do
          ActiveRecord::Base.cache { yield }
        end
        log notice("Completed #{message.type.inspect} event in #{timing} seconds")
      end

      def decode(payload)
        MultiJson.decode(payload)
      end

      def connection
        @connection ||= AMQP.start(config)
      end

      def channel
        @channel ||=  AMQP::Channel.new(connection).prefetch(config.prefetch)
      end
<commit_msg>cant use a var if i cant use it :)<commit_after>require 'amqp'
require 'multi_json'
require 'hashr'
require 'benchmark'

module Travis
  class Consumer
    autoload :Handler, 'travis/consumer/handler'
    autoload :Job,     'travis/consumer/job'
    autoload :Worker,  'travis/consumer/worker'

    include Logging

    REPORTING_KEY = 'reporting.jobs'

    class << self
      def start(options = {})
        Database.connect(options)

        EM.run do
          prune_workers!
          # cleanup_jobs
          subscribe
        end
      end

      def subscribe
        new.subscribe
      end

      def prune_workers!
        interval = Travis.config.workers.prune.interval
        EM.add_periodic_timer(interval, &::Worker.method(:prune))
      end

      def cleanup_jobs!
        interval = Travis.config.jobs.retry.interval
        EM.add_periodic_timer(interval, &::Job.method(:cleanup))
      end
    end

    attr_reader :config

    def initialize
      @config = Travis.config.amqp
    end

    def subscribe
      queue.subscribe(:ack => true, &method(:receive))
    end

    def receive(message, payload)
      log notice("Handling event #{message.type.inspect} with payload : #{payload.inspect}")

      event   = message.type
      payload = decode(payload)
      handler = handler_for(event, payload)

      benchmark_and_cache do
        handler.handle
      end

      message.ack
    rescue Exception => e
      puts e.message, e.backtrace
      message.ack
      # message.reject(:requeue => false) # how to decide whether to requeue the message?
    end

    def heartbeat(message, payload)
      log notice("Heartbeat: #{payload.inspect}")
      Worker.heartbeat
    end

    protected

      def handler_for(event, payload)
        case event.to_s
        when /^job/
          Job.new(event, payload)
        when /^worker/
          Worker.new(event, payload)
        else
          raise "Unknown message type: #{event.inspect}"
        end
      end

      def benchmark_and_cache
        timing = Benchmark.realtime do
          ActiveRecord::Base.cache { yield }
        end
        log notice("Completed in #{timing} seconds")
      end

      def decode(payload)
        MultiJson.decode(payload)
      end

      def connection
        @connection ||= AMQP.start(config)
      end

      def channel
        @channel ||=  AMQP::Channel.new(connection).prefetch(config.prefetch)
      end

{'fraction_non_alphanumeric': 0.09072207364739765, 'fraction_numerical': 0.0, 'mean_word_length': 2.140826873385013, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2094109', 'n_tokens_mistral': 1456, 'n_tokens_neox': 1377, 'n_words': 345}