<commit_before>module QueueStick
  class Worker
    # TODO(dbalatero): configurable?
    MAX_ERRORS = 50

    attr_reader :errors

    @@counters = Hash.new { |h, k| h[k] = [] }

    def initialize
      @errors = []
      create_counters!
    end

    def get_message_from_queue
      raise NotImplementedError, "#get_message_from_queue needs to be implemented in a subclass."
    end

    def delete_message_from_queue(message)
      raise NotImplementedError, "#delete_message_from_queue needs to be implemented in a subclass."
    end

    def run_loop
      # Exit the run loop if we get a shutdown request.
      Thread.current.exit if Thread.current[:shutdown]

      message = get_message_from_queue

      begin
        process(message.body)
        delete_message_from_queue(message)
        counter(:messages_processed).increment!
      rescue Exception => process_error
        error = WorkerError.new(message.id)
        error.exceptions << process_error
        begin
          recover
        rescue Exception => recover_error
          error.exceptions << recover_error
        end
        @errors << error
        @errors.shift if @errors.size > MAX_ERRORS
      end<commit_msg>Clarified some expectations of get_message_from_queue.<commit_after>module QueueStick
  class Worker
    # TODO(dbalatero): configurable?
    MAX_ERRORS = 50

    attr_reader :errors

    @@counters = Hash.new { |h, k| h[k] = [] }

    def initialize
      @errors = []
      create_counters!
    end

    # This method needs to return either:
    # a) a Message object if there is something in the queue
    # b) nil if there is nothing in the queue
    def get_message_from_queue
      raise NotImplementedError, "#get_message_from_queue needs to be implemented in a subclass."
    end

    def delete_message_from_queue(message)
      raise NotImplementedError, "#delete_message_from_queue needs to be implemented in a subclass."
    end

    def run_loop
      # Exit the run loop if we get a shutdown request.
      Thread.current.exit if Thread.current[:shutdown]

      message = get_message_from_queue

      begin
        process(message.body)
        delete_message_from_queue(message)
        counter(:messages_processed).increment!
      rescue Exception => process_error
        error = WorkerError.new(message.id)
        error.exceptions << process_error
        begin
          recover
        rescue Exception => recover_error
          error.exceptions << recover_error
        end
        @errors << error
        @errors.shift if @errors.size > MAX_ERRORS
      end
{'fraction_non_alphanumeric': 0.06705426356589148, 'fraction_numerical': 0.0015503875968992248, 'mean_word_length': 2.599721059972106, 'pattern_counts': {'":': 0, '<': 15, '<?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': '2100711', 'n_tokens_mistral': 741, 'n_tokens_neox': 714, 'n_words': 207}