<commit_before>require 'multi_json'

require 'travis'
require 'travis/support'
# require 'travis/log_subscriber/active_record_metrics'
require 'timeout'
require 'memory'

$stdout.sync = true

module Travis
  class Logs
    autoload :Handler, 'travis/logs/handler'

    extend Exceptions::Handling
    include Logging

    class << self
      def start
        setup
        new.subscribe
      end

      protected

        def setup
          Memory.report_periodically

          Travis::Async.enabled = true
          Travis::Database.connect
          Travis::Exceptions::Reporter.start
          Travis::Notification.setup
          Travis::Amqp.config = Travis.config.amqp
          # Travis::Features.start
          # Travis::LogSubscriber::ActiveRecordMetrics.attach

          NewRelic.start if File.exists?('config/newrelic.yml')
        end
    end

    def subscribe
      info 'Subscribing to amqp ...'
      info "Subscribing to reporting.jobs.logs"

      Travis::Amqp::Consumer.jobs('logs').subscribe(:ack => true) do |msg, payload|
        receive(:route, msg, payload)
      end

      0.upto(Travis.config.logs.shards - 1).each do |shard|
        info "Subscribing to reporting.jobs.logs.#{shard}"
        Travis::Amqp::Consumer.jobs("logs.#{shard}").subscribe(:ack => true) do |msg, payload|
          receive(:log, msg, payload)
        end
      end
    end

    def receive(type, message, payload)
      return unless payload = decode(payload)
      Travis.uuid = payload['uuid']
      handle(type, payload)
    rescue Exception => e
      puts "!!!FAILSAFE!!! #{e.message}", e.backtrace
    ensure
      message.ack
    end

    protected

      def handle(type, payload)
        timeout do
          Travis::Logs::Handler.handle(type, payload)
        end
      end
      rescues :handle, :from => Exception unless Travis.env == 'test'

      def timeout(&block)
        Timeout::timeout(60, &block)
      end

      def decode(payload)
        MultiJson.decode(payload)
      rescue StandardError => e
        error "[#{Thread.current.object_id}] [decode error] payload could not be decoded with engine #{MultiJson.engine.to_s} (#{e.message}): #{payload.inspect}"
        nil
      end
  end
end
<commit_msg>try setting channel.prefetch to number of shards + 1<commit_after>require 'multi_json'

require 'travis'
require 'travis/support'
# require 'travis/log_subscriber/active_record_metrics'
require 'timeout'
require 'memory'

$stdout.sync = true

module Travis
  class Logs
    autoload :Handler, 'travis/logs/handler'

    extend Exceptions::Handling
    include Logging

    class << self
      def start
        setup
        new.subscribe
      end

      protected

        def setup
          Memory.report_periodically

          Travis::Async.enabled = true
          Travis::Database.connect
          Travis::Exceptions::Reporter.start
          Travis::Notification.setup

          Travis::Amqp.config = Travis.config.amqp
          Travis::Amqp.channel.prefetch = Travis.config.logs.shards + 1

          # Travis::Features.start
          # Travis::LogSubscriber::ActiveRecordMetrics.attach

          NewRelic.start if File.exists?('config/newrelic.yml')
        end
    end

    def subscribe
      info 'Subscribing to amqp ...'
      info "Subscribing to reporting.jobs.logs"

      Travis::Amqp::Consumer.jobs('logs').subscribe(:ack => true) do |msg, payload|
        receive(:route, msg, payload)
      end

      0.upto(Travis.config.logs.shards - 1).each do |shard|
        info "Subscribing to reporting.jobs.logs.#{shard}"
        Travis::Amqp::Consumer.jobs("logs.#{shard}").subscribe(:ack => true) do |msg, payload|
          receive(:log, msg, payload)
        end
      end
    end

    def receive(type, message, payload)
      return unless payload = decode(payload)
      Travis.uuid = payload['uuid']
      handle(type, payload)
    rescue Exception => e
      puts "!!!FAILSAFE!!! #{e.message}", e.backtrace
    ensure
      message.ack
    end

    protected

      def handle(type, payload)
        timeout do
          Travis::Logs::Handler.handle(type, payload)
        end
      end
      rescues :handle, :from => Exception unless Travis.env == 'test'

      def timeout(&block)
        Timeout::timeout(60, &block)
      end

      def decode(payload)
        MultiJson.decode(payload)
      rescue StandardError => e
        error "[#{Thread.current.object_id}] [decode error] payload could not be decoded with engine #{MultiJson.engine.to_s} (#{e.message}): #{payload.inspect}"
        nil
      end
  end
end

{'fraction_non_alphanumeric': 0.11062622736198996, 'fraction_numerical': 0.002181976871045167, 'mean_word_length': 2.4466165413533836, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, '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': '872885', 'n_tokens_mistral': 1375, 'n_tokens_neox': 1354, 'n_words': 322}