<commit_before>module Bandersnatch
  class Subscriber < Base

    attr_accessor :handlers

    EXCHANGE_CREATION_KEYS  = [:auto_delete, :durable, :internal, :nowait, :passive]
    RECOVER_AFTER           = 10.seconds

    def initialize(client, options = {})
      super
      @handlers = {}
      @amqp_connections = {}
      @mqs = {}
    end

    def listen(messages=@messages.keys)
      EM.run do
        create_exchanges(messages)
        bind_queues(messages)
        subscribe
      end
    end

    def register_handler(messages, opts, &block)
      Array(messages).each do |message|
        (@handlers[message] ||= []) << [opts.symbolize_keys, block]
      end
    end

    private

    def subscribe(messages=nil)
      messages ||= @messages.keys
      Array(messages).each do |message|
        @servers.each do |s|
          set_current_server s
          subscribe_message(message)
        end
      end
    end

    def queues_with_handlers(messages)
      messages.map do |name|
        @handlers[name].map {|opts, _| opts[:queue] || name }
      end.flatten
    end

    def mq
      @mqs[@server] ||= MQ.new(amqp_connection)
    end

    def subscribe_message(message)
      handlers = Array(@handlers[message])
      error("no handler for message #{message}") if handlers.empty?
      handlers.each do |opts, block|
        opts = opts.dup
        key = opts.delete(:key) || message
        queue = opts.delete(:queue) || message
        callback = create_subscription_callback(@server, queue, block)
        logger.debug "subscribing to queue #{queue} with key #{key} for message #{message}"
        begin
          queues[queue].subscribe(opts.merge(:key => "#{key}.#"), &callback)
        rescue MQ::Error
          error("Binding multiple handlers for the same queue isn't possible. You might want to use the :queue option")
        end
      end
    end

    def create_subscription_callback(server, queue, block)
      lambda do |header,data|
        begin
          m = Message.new(queue, header, data)
          m.server = server
          m.process(block)
        rescue Exception
          logger.error "Error during message processing. Message will get redelivered. #{m}\n #{$!}"
          @timer.cancel if @timer
          @timer = EM::Timer.new(RECOVER_AFTER) do
            logger.info "Redelivering unacked messages that could not be verified because of unavailable Redis"
            @mqs[server].recover(true)
          end
        end
      end
    end

    def create_exchange!(name, opts)
      mq.__send__(opts[:type], name, opts.slice(*EXCHANGE_CREATION_KEYS))
    end

    def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys)
      queue = mq.queue(queue_name, creation_keys)
      queue.bind(exchange(exchange_name), binding_keys)
      queue
    end

    def stop!
      @amqp_connections.each_value{|c| c.close}
      @amqp_connections = {}
      @mqs = {}
      EM.stop_event_loop
    end

    def new_amqp_connection
      AMQP.connect(:host => current_host, :port => current_port)
    end<commit_msg>extracted method<commit_after>module Bandersnatch
  class Subscriber < Base

    attr_accessor :handlers

    EXCHANGE_CREATION_KEYS  = [:auto_delete, :durable, :internal, :nowait, :passive]
    RECOVER_AFTER           = 10.seconds

    def initialize(client, options = {})
      super
      @handlers = {}
      @amqp_connections = {}
      @mqs = {}
    end

    def listen(messages=@messages.keys)
      EM.run do
        create_exchanges(messages)
        bind_queues(messages)
        subscribe
      end
    end

    def register_handler(messages, opts, &block)
      Array(messages).each do |message|
        (@handlers[message] ||= []) << [opts.symbolize_keys, block]
      end
    end

    private

    def subscribe(messages=nil)
      messages ||= @messages.keys
      Array(messages).each do |message|
        @servers.each do |s|
          set_current_server s
          subscribe_message(message)
        end
      end
    end

    def queues_with_handlers(messages)
      messages.map do |name|
        @handlers[name].map {|opts, _| opts[:queue] || name }
      end.flatten
    end

    def mq
      @mqs[@server] ||= MQ.new(amqp_connection)
    end

    def subscribe_message(message)
      handlers = Array(@handlers[message])
      error("no handler for message #{message}") if handlers.empty?
      handlers.each do |opts, block|
        opts = opts.dup
        key = opts.delete(:key) || message
        queue = opts.delete(:queue) || message
        callback = create_subscription_callback(@server, queue, block)
        logger.debug "subscribing to queue #{queue} with key #{key} for message #{message}"
        begin
          queues[queue].subscribe(opts.merge(:key => "#{key}.#"), &callback)
        rescue MQ::Error
          error("Binding multiple handlers for the same queue isn't possible. You might want to use the :queue option")
        end
      end
    end

    def create_subscription_callback(server, queue, block)
      lambda do |header,data|
        begin
          m = Message.new(queue, header, data)
          m.server = server
          m.process(block)
        rescue Exception
          logger.error "Error during message processing. Message will get redelivered. #{m}\n #{$!}"
          install_recovery_timer(server)
        end
      end
    end

    def install_recovery_timer(server)
      @timer.cancel if @timer
      @timer = EM::Timer.new(RECOVER_AFTER) do
        logger.info "Redelivering unacked messages that could not be verified because of unavailable Redis"
        @mqs[server].recover(true)
      end
    end

    def create_exchange!(name, opts)
      mq.__send__(opts[:type], name, opts.slice(*EXCHANGE_CREATION_KEYS))
    end

    def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys)
      queue = mq.queue(queue_name, creation_keys)
      queue.bind(exchange(exchange_name), binding_keys)
      queue
    end

    def stop!
      @amqp_connections.each_value{|c| c.close}
      @amqp_connections = {}
      @mqs = {}
      EM.stop_event_loop
    end

    def new_amqp_connection
      AMQP.connect(:host => current_host, :port => current_port)
    end
{'fraction_non_alphanumeric': 0.102803738317757, 'fraction_numerical': 0.0006445375443119562, 'mean_word_length': 2.4540901502504173, 'pattern_counts': {'":': 0, '<': 9, '<?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': '4768750', 'n_tokens_mistral': 1892, 'n_tokens_neox': 1832, 'n_words': 456}