<commit_before>module BaseListener
  class Listener
    INTERFACE = %w(queue routing_key)
    attr_reader :appid, :logger, :connection

    def initialize(appid)
      @appid = appid
      @logger = Logger.new self
      @connection = Connection.new @logger
      logger.info "Initialize new #{log_name} with appid = #{appid}"
    end

    def subscribe!
      queue.bind(exchange, routing_key: routing_key).subscribe(block: true) do |info, meta, payload|
        perform info, meta, Marshal.load(payload)
      end
    end

    def log_name
      self.class.name
    end

    private

    def perform(info, meta, payload)
      logger.info "Message with payload: #{payload.inspect} received"
      requeue_if_needed(payload) { handle_errors { worker_for(payload).perform } }
    end

    def worker_for(payload)
      payload[:worker].split('::').inject Object do |namespace, name|
        namespace.const_get name
      end.new payload[:message]
    end

    def requeue(payload)
      payload[:requeue_tries] ||= 0
      if payload[:requeue_tries] <=  max_requeue_tries
        payload[:requeue_tries] +=  1
        logger.warn "requeue message with payload: #{payload.inspect}"
        exchange.publish Marshal.dump(payload), routing_key: "#{Config.prefix}routing_keys.retry.#{appid}", persisted: true
      else
        logger.error "message with payload: #{payload.inspect} can't be requeued anymore"
      end
    end

    def max_requeue_tries
      3
    end

    def requeue_if_needed(payload)
      requeue(payload) unless yield
    end

    #META interface
    INTERFACE.each do |name|
      define_method(name) { raise MustBeDefinedInChildClass }
    end

    #META def channel and def connection
    %w(channel exchange).each do |name|
      define_method(name) { connection.public_send name }
    end

    def handle_errors
      yield
    rescue => e
      logger.error e.message
      Config.handler.handle e, self.class.name
      false
    end
  end
end<commit_msg>issue with rescue was fixed<commit_after>module BaseListener
  class Listener
    INTERFACE = %w(queue routing_key)
    attr_reader :appid, :logger, :connection

    def initialize(appid)
      @appid = appid
      @logger = Logger.new self
      @connection = Connection.new @logger
      logger.info "Initialize new #{log_name} with appid = #{appid}"
    end

    def subscribe!
      queue.bind(exchange, routing_key: routing_key).subscribe(block: true) do |info, meta, payload|
        perform info, meta, Marshal.load(payload)
      end
    end

    def log_name
      self.class.name
    end

    private

    def perform(info, meta, payload)
      logger.info "Message with payload: #{payload.inspect} received"
      requeue_if_needed(payload) { handle_errors { worker_for(payload).perform } }
    end

    def worker_for(payload)
      payload[:worker].split('::').inject Object do |namespace, name|
        namespace.const_get name
      end.new payload[:message]
    end

    def requeue(payload)
      payload[:requeue_tries] ||= 0
      if payload[:requeue_tries] <=  max_requeue_tries
        payload[:requeue_tries] +=  1
        logger.warn "requeue message with payload: #{payload.inspect}"
        exchange.publish Marshal.dump(payload), routing_key: "#{Config.prefix}routing_keys.retry.#{appid}", persisted: true
      else
        logger.error "message with payload: #{payload.inspect} can't be requeued anymore"
      end
    end

    def max_requeue_tries
      3
    end

    def requeue_if_needed(payload)
      requeue(payload) unless yield
    end

    #META interface
    INTERFACE.each do |name|
      define_method(name) { raise MustBeDefinedInChildClass }
    end

    #META def channel and def connection
    %w(channel exchange).each do |name|
      define_method(name) { connection.public_send name }
    end

    def handle_errors
      yield
    rescue Exception => e
      logger.error e.message
      Config.handler.handle e, self.class.name
      false
    end
  end
end
{'fraction_non_alphanumeric': 0.09168336673346693, 'fraction_numerical': 0.001503006012024048, 'mean_word_length': 2.810114503816794, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '2619747', 'n_tokens_mistral': 1224, 'n_tokens_neox': 1187, 'n_words': 316}