<commit_before>require "barrister/amqp/version"
require 'bunny'
require 'barrister'
require 'ostruct'

module Barrister
  module Amqp
    Wrapper = Struct.new(:message) do
      def wrap
        {
          "id"      => Barrister.rand_str(22),
          "message" => JSON.generate(message, { :ascii_only=>true })
        }
      end

      def unwrap
        message
      end
    end

    Config = OpenStruct.new(debug: false, wrapper: Wrapper)

    class Transport
      # NOTE: transport needs to implement request method for the Barrister::Client to
      # send requests to the server
      def initialize(service_name, options={})
        conn = Bunny.new ENV.fetch('AMQP_URL') #"Please set AMQP_URL to something like: amqp://user:password@host:port/vhost"
        conn.start
        @ch             = conn.create_channel
        @service_q      = @ch.queue(service_name, auto_delete: false)
        @reply_q        = @ch.queue('', exclusive: true)
        @x              = @ch.default_exchange
        @response_table = Hash.new { |h,k| h[k] = Queue.new }

        @reply_q.subscribe(block: false) do |delivery_info, properties, payload|
          @response_table[properties[:correlation_id]].push payload # push anything that comes in the response_q
        end
      end

      def request(message)
        enveloppe = Config.wrapper.new(message).wrap
        # NOTE message could be an array
        print "[AMQP TRANSPORT --->] \n #{enveloppe} \n" if Config.debug
        @x.publish(enveloppe['message'], { correlation_id: enveloppe['id'], reply_to: @reply_q.name, routing_key: @service_q.name})

        response = @response_table[enveloppe['id']].pop
        @response_table.delete enveloppe['id']

        begin
          JSON.parse(response).tap do |resp|
            print "[AMQP TRANSPORT <---] \n #{resp} \n" if Config.debug
          end<commit_msg>Adds the timeout option to the AMQP transport.<commit_after>require "barrister/amqp/version"
require 'bunny'
require 'barrister'
require 'ostruct'

module Barrister
  module Amqp
    Wrapper = Struct.new(:message) do
      def wrap
        {
          "id"      => Barrister.rand_str(22),
          "message" => JSON.generate(message, { :ascii_only=>true })
        }
      end

      def unwrap
        message
      end
    end

    Config = OpenStruct.new(debug: false, wrapper: Wrapper)

    class Transport
      # NOTE: transport needs to implement request method for the Barrister::Client to
      # send requests to the server
      def initialize(service_name, options={})
        conn = Bunny.new ENV.fetch('AMQP_URL') #"Please set AMQP_URL to something like: amqp://user:password@host:port/vhost"
        conn.start
        @ch             = conn.create_channel
        @service_q      = @ch.queue(service_name, auto_delete: false)
        @reply_q        = @ch.queue('', exclusive: true)
        @x              = @ch.default_exchange
        @response_table = Hash.new { |h,k| h[k] = Queue.new }
        @timeout        = options[:timeout] || 300

        @reply_q.subscribe(block: false) do |delivery_info, properties, payload|
          @response_table[properties[:correlation_id]].push payload # push anything that comes in the response_q
        end
      end

      def request(message)
        enveloppe = Config.wrapper.new(message).wrap
        # NOTE message could be an array
        print "[AMQP TRANSPORT --->] \n #{enveloppe} \n" if Config.debug
        @x.publish(enveloppe['message'], { correlation_id: enveloppe['id'], reply_to: @reply_q.name, routing_key: @service_q.name})

        response = Timeout::timeout(@timeout) do
          response = @response_table[enveloppe['id']].pop
          @response_table.delete enveloppe['id']
          response
        end

        begin
          JSON.parse(response).tap do |resp|
            print "[AMQP TRANSPORT <---] \n #{resp} \n" if Config.debug
          end
{'fraction_non_alphanumeric': 0.11315924219150025, 'fraction_numerical': 0.0017921146953405018, 'mean_word_length': 2.388551604509974, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '6472435', 'n_tokens_mistral': 1194, 'n_tokens_neox': 1139, 'n_words': 296}