<commit_before>require 'aws/sqs'

module Chore
  class SQSConsumer < Consumer

    def initialize(queue_name, opts={})
      super(queue_name, opts)
      @sqs = AWS::SQS.new(
        :access_key_id => Chore.config.aws_access_key,
        :secret_access_key => Chore.config.aws_secret_key)
      @queue = @sqs.queues.named(@queue_name)
      @dupes = DuplicateDetector.new(Chore.config.dedupe_servers || nil)
    end

    def consume(&handler)
      # this is for spec purposes, so we can test this w/out looping forever
      while running?
        begin
          msg = @queue.receive_messages(:limit => 10)
          next if msg.nil? || msg.empty?

          handle_messages(*msg, &handler)
        rescue => e
          Chore.logger.error { "SQSConsumer#Consume: #{e.inspect}" }
        end
      end
    end

    def reject(id)

    end

    def complete(id)
      Chore.logger.debug "Completing (deleting): #{id}"
      @queue.batch_delete([id])
    end

    private

    def handle_messages(*messages, &block)
      messages.each do |message|
        block.call(message.handle, message.body) unless duplicate_message?(message)
        Chore.run_hooks_for(:on_fetch, message)
      end
    end

    def duplicate_message?(message)
      @dupes.found_duplicate?(message)
    end
  end
end
<commit_msg>whoops forgot to update the hooks in sqs runner<commit_after>require 'aws/sqs'

module Chore
  class SQSConsumer < Consumer

    def initialize(queue_name, opts={})
      super(queue_name, opts)
      @sqs = AWS::SQS.new(
        :access_key_id => Chore.config.aws_access_key,
        :secret_access_key => Chore.config.aws_secret_key)
      @queue = @sqs.queues.named(@queue_name)
      @dupes = DuplicateDetector.new(Chore.config.dedupe_servers || nil)
    end

    def consume(&handler)
      # this is for spec purposes, so we can test this w/out looping forever
      while running?
        begin
          msg = @queue.receive_messages(:limit => 10)
          next if msg.nil? || msg.empty?

          handle_messages(*msg, &handler)
        rescue => e
          Chore.logger.error { "SQSConsumer#Consume: #{e.inspect}" }
        end
      end
    end

    def reject(id)

    end

    def complete(id)
      Chore.logger.debug "Completing (deleting): #{id}"
      @queue.batch_delete([id])
    end

    private

    def handle_messages(*messages, &block)
      messages.each do |message|
        block.call(message.handle, message.body) unless duplicate_message?(message)
        Chore.run_hooks_for(:on_fetch, *message['args'])
      end
    end

    def duplicate_message?(message)
      @dupes.found_duplicate?(message)
    end
  end
end

{'fraction_non_alphanumeric': 0.10809792843691149, 'fraction_numerical': 0.0015065913370998117, 'mean_word_length': 2.5843454790823213, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '347501', 'n_tokens_mistral': 870, 'n_tokens_neox': 853, 'n_words': 190}