<commit_before>require 'bson'
require 'json'
require 'multi_json'
require 'amqp'
require 'goliath'

module Rodent
  module Goliath
    class Middleware
      include ::Goliath::Rack::AsyncMiddleware

      def initialize(app, header_proxy_name = 'Rodent-Proxy')
        @header_proxy_name = header_proxy_name
        super(app)
      end

      def hook_into_callback_chain(env, *args)
        async_callback = env['async.callback']

        downstream_callback = proc do |status, headers, response|
          result_response = [status, headers, response]
          if proxy_type = headers.delete(@header_proxy_name)
            body = response.respond_to?(:body) ? response.body.join : response
            result_response = safely(env) { proxy_request(env, proxy_type, body, headers) }
          end
          async_callback.call(result_response) unless result_response == ::Goliath::Connection::AsyncResponse
        end

        env['async.callback'] = downstream_callback
      end

      protected
      def proxy_request(env, type, body, headers = {})
        async_callback = env['async.callback']

        message_id = BSON::ObjectId.new.to_s

        env.channels.execute(false) do |channel|
          replies_queue = channel.queue(message_id, exclusive: true, auto_delete: true)

          consumer = AMQP::Consumer.new(channel, replies_queue)
          bind_consumer(consumer, async_callback, headers)

          channel.direct('rodent.requests').publish(body, routing_key: type, message_id: message_id, reply_to: replies_queue.name)
        end

        ::Goliath::Connection::AsyncResponse
      end

      def bind_consumer(consumer, async_callback, headers)
        consumer.consume do
          consumer.on_delivery do |metadata, payload|
            response = MultiJson.load(payload)
            response['headers']['Content-Length'] = response['body'].length.to_s
            response['headers']['Content-Type'] = 'application/json'
            async_callback.call([response['status'], headers.merge(response['headers']), response['body']])
            metadata.ack
            consumer.cancel
          end
        end
      end
    end
  end
end<commit_msg>Fixed content length<commit_after>require 'bson'
require 'json'
require 'multi_json'
require 'amqp'
require 'goliath'

module Rodent
  module Goliath
    class Middleware
      include ::Goliath::Rack::AsyncMiddleware

      def initialize(app, header_proxy_name = 'Rodent-Proxy')
        @header_proxy_name = header_proxy_name
        super(app)
      end

      def hook_into_callback_chain(env, *args)
        async_callback = env['async.callback']

        downstream_callback = proc do |status, headers, response|
          result_response = [status, headers, response]
          if proxy_type = headers.delete(@header_proxy_name)
            body = response.respond_to?(:body) ? response.body.join : response
            result_response = safely(env) { proxy_request(env, proxy_type, body, headers) }
          end
          async_callback.call(result_response) unless result_response == ::Goliath::Connection::AsyncResponse
        end

        env['async.callback'] = downstream_callback
      end

      protected
      def proxy_request(env, type, body, headers = {})
        async_callback = env['async.callback']

        message_id = BSON::ObjectId.new.to_s

        env.channels.execute(false) do |channel|
          replies_queue = channel.queue(message_id, exclusive: true, auto_delete: true)

          consumer = AMQP::Consumer.new(channel, replies_queue)
          bind_consumer(consumer, async_callback, headers)

          channel.direct('rodent.requests').publish(body, routing_key: type, message_id: message_id, reply_to: replies_queue.name)
        end

        ::Goliath::Connection::AsyncResponse
      end

      def bind_consumer(consumer, async_callback, headers)
        consumer.consume do
          consumer.on_delivery do |metadata, payload|
            response = MultiJson.load(payload)
            response['headers']['Content-Length'] = response['body'].to_s.bytes.length.to_s
            response['headers']['Content-Type'] = 'application/json'
            async_callback.call([response['status'], headers.merge(response['headers']), response['body']])
            metadata.ack
            consumer.cancel
          end
        end
      end
    end
  end
end
{'fraction_non_alphanumeric': 0.10283363802559416, 'fraction_numerical': 0.0, 'mean_word_length': 2.8768821966341895, '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': '6697175', 'n_tokens_mistral': 1225, 'n_tokens_neox': 1188, 'n_words': 251}