<commit_before>class Freddy
  module Producers
    class SendAndWaitResponseProducer
      NO_ROUTE = 312
      CONTENT_TYPE = 'application/json'.freeze

      def initialize(channel, logger)
        @logger = logger
        @channel = channel

        @request_manager = RequestManager.new(@logger)

        @exchange = @channel.default_exchange
        @topic_exchange = @channel.topic Freddy::FREDDY_TOPIC_EXCHANGE_NAME

        @channel.on_return do |reply_code, correlation_id|
          if reply_code == NO_ROUTE
            @request_manager.no_route(correlation_id)
          end
        end

        @response_queue = @channel.queue("", exclusive: true)
        @request_manager.start

        @response_consumer = Consumers::ResponseConsumer.new(@logger)
        @response_consumer.consume(@response_queue, &method(:handle_response))
      end

      def produce(destination, payload, properties)
        timeout_seconds = properties.fetch(:timeout)
        container = SyncResponseContainer.new
        async_request destination, payload, properties, &container
        container.wait_for_response(timeout_seconds + 0.1)
      end

      private

      def async_request(destination, payload, properties, &block)
        timeout = properties.fetch(:timeout)
        delete_on_timeout = properties.fetch(:delete_on_timeout)
        properties.delete(:timeout)
        properties.delete(:delete_on_timeout)

        correlation_id = SecureRandom.uuid
        @request_manager.store(correlation_id, callback: block, destination: destination, timeout: Time.now + timeout)

        if delete_on_timeout
          properties[:expiration] = (timeout * 1000).to_i
        end

        properties = properties.merge(
          routing_key: destination, content_type: CONTENT_TYPE,
          correlation_id: correlation_id, reply_to: @response_queue.name,
          mandatory: true, type: 'request'
        )
        json_payload = Payload.dump(payload)

        @logger.debug "Publishing request with payload #{payload.inspect} to #{destination}, waiting for response on #{@response_queue.name} with correlation_id #{correlation_id}"

        # Connection adepters handle thread safety for #publish themselves. No
        # need to lock these.
        @topic_exchange.publish json_payload, properties.dup
        @exchange.publish json_payload, properties.dup
      end

      def handle_response(delivery)
        correlation_id = delivery.correlation_id<commit_msg>Use hash destruction for better interface<commit_after>class Freddy
  module Producers
    class SendAndWaitResponseProducer
      NO_ROUTE = 312
      CONTENT_TYPE = 'application/json'.freeze

      def initialize(channel, logger)
        @logger = logger
        @channel = channel

        @request_manager = RequestManager.new(@logger)

        @exchange = @channel.default_exchange
        @topic_exchange = @channel.topic Freddy::FREDDY_TOPIC_EXCHANGE_NAME

        @channel.on_return do |reply_code, correlation_id|
          if reply_code == NO_ROUTE
            @request_manager.no_route(correlation_id)
          end
        end

        @response_queue = @channel.queue("", exclusive: true)
        @request_manager.start

        @response_consumer = Consumers::ResponseConsumer.new(@logger)
        @response_consumer.consume(@response_queue, &method(:handle_response))
      end

      def produce(destination, payload, properties)
        timeout_seconds = properties.fetch(:timeout)
        container = SyncResponseContainer.new
        async_request destination, payload, properties, &container
        container.wait_for_response(timeout_seconds + 0.1)
      end

      private

      def async_request(destination, payload, timeout:, delete_on_timeout:, **properties, &block)
        correlation_id = SecureRandom.uuid
        @request_manager.store(correlation_id, callback: block, destination: destination, timeout: Time.now + timeout)

        if delete_on_timeout
          properties[:expiration] = (timeout * 1000).to_i
        end

        properties = properties.merge(
          routing_key: destination, content_type: CONTENT_TYPE,
          correlation_id: correlation_id, reply_to: @response_queue.name,
          mandatory: true, type: 'request'
        )
        json_payload = Payload.dump(payload)

        @logger.debug "Publishing request with payload #{payload.inspect} to #{destination}, waiting for response on #{@response_queue.name} with correlation_id #{correlation_id}"

        # Connection adapters handle thread safety for #publish themselves. No
        # need to lock these.
        @topic_exchange.publish json_payload, properties.dup
        @exchange.publish json_payload, properties.dup
      end

      def handle_response(delivery)
        correlation_id = delivery.correlation_id
{'fraction_non_alphanumeric': 0.08036643764313971, 'fraction_numerical': 0.0037476577139287947, 'mean_word_length': 3.0403700588730027, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5709301', 'n_tokens_mistral': 1304, 'n_tokens_neox': 1229, 'n_words': 288}