<commit_before>module Liebre
  class Publisher
    
    def initialize publisher_name
      @publisher_name = publisher_name
    end
    
    def enqueue message, options = {}
      with_connection do
        logger.debug "Liebre: Publishing '#{message}' with '#{options}' to exchange: #{exchange.name}"
        exchange.publish message, options
      end
    end
    
    def enqueue_and_wait message, options = {}
      result = nil
      with_connection do
        correlation_id = options[:correlation_id] ||= generate_uuid
        reply_queue = reply_queue correlation_id
        options[:reply_to] = reply_queue.name
        reply_queue.subscribe(:block => false) do |delivery_info, meta, payload|
          if meta[:correlation_id] == correlation_id
            result = payload
            logger.debug "Liebre: Received response '#{result}'"
            channel.consumers[delivery_info.consumer_tag].cancel
          end
        end
        logger.debug "Liebre: Publishing '#{message}' with '#{options}' to exchange: #{exchange.name}"
        exchange.publish message, options
        begin
          Timeout.timeout(Liebre.config.rpc_request_timeout) do
            sleep 0.01 while result.nil?
          end
        rescue Timeout::Error
          #do nothing
        ensure
          begin
            reply_queue.delete
          rescue Timeout::Error
            logger.error "error while trying to delete RPC exclusive queue"
          end
        end
      end
      result
    end
    
    alias_method :rpc, :enqueue_and_wait
    
    private
    
    def with_connection
      connection_manager.ensure_started
      yield
    end
    
    def reply_queue correlation_id
      queue_name = "#{publisher_name}_callback_#{correlation_id}"
      channel.queue queue_name, :exclusive => true
    end
    
    def exchange
      @exchange ||= Liebre::Common::Utils.create_exchange channel, exchange_config
    end
    
    def channel
      @channel ||= connection_manager.channel_for(connection_name)
    end
    
    def publishers
      Liebre.config.publishers
    end

    def exchange_config
      config.fetch("exchange")
    end
    <commit_msg>check for closed connection exception and retry<commit_after>module Liebre
  class Publisher
    
    def initialize publisher_name
      @publisher_name = publisher_name
    end
    
    def enqueue message, options = {}
      with_connection do
        logger.debug "Liebre: Publishing '#{message}' with '#{options}' to exchange: #{exchange.name}"
        exchange.publish message, options
      end
    end
    
    def enqueue_and_wait message, options = {}
      result = nil
      with_connection do
        correlation_id = options[:correlation_id] ||= generate_uuid
        reply_queue = reply_queue correlation_id
        options[:reply_to] = reply_queue.name
        reply_queue.subscribe(:block => false) do |delivery_info, meta, payload|
          if meta[:correlation_id] == correlation_id
            result = payload
            logger.debug "Liebre: Received response '#{result}'"
            channel.consumers[delivery_info.consumer_tag].cancel
          end
        end
        logger.debug "Liebre: Publishing '#{message}' with '#{options}' to exchange: #{exchange.name}"
        exchange.publish message, options
        begin
          Timeout.timeout(Liebre.config.rpc_request_timeout) do
            sleep 0.01 while result.nil?
          end
        rescue Timeout::Error
          #do nothing
        ensure
          begin
            reply_queue.delete
          rescue Timeout::Error
            logger.error "error while trying to delete RPC exclusive queue"
          end
        end
      end
      result
    end
    
    alias_method :rpc, :enqueue_and_wait
    
    private
    
    def with_connection
      connection_manager.ensure_started
      begin
        yield
      rescue ConnectionClosedError
        logger.warn("#{self.class.name}: ConnectionClosedError found restarting connection")
        connection_manager.restart
        retry
      end
    end
    
    def reply_queue correlation_id
      queue_name = "#{publisher_name}_callback_#{correlation_id}"
      channel.queue queue_name, :exclusive => true
    end
    
    def exchange
      @exchange ||= Liebre::Common::Utils.create_exchange channel, exchange_config
    end
    
    def channel
      @channel ||= connection_manager.channel_for(connection_name)
    end
    
    def publishers
      Liebre.config.publishers
    end

    def exchange_config
      config.fetch("exchange")
    end
    
{'fraction_non_alphanumeric': 0.07175672719317436, 'fraction_numerical': 0.0013126230584117262, 'mean_word_length': 2.237960339943343, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '7114092', 'n_tokens_mistral': 1229, 'n_tokens_neox': 1196, 'n_words': 311}