<commit_before>module Combi
  class Queue < Bus

    def post_initialize
      unless @options[:init_queue] == false
        require 'queue_service'
        EventMachine.next_tick do
          QueueService.start(ConfigProvider.for(:amqp), rpc: :enabled)
        end
      end
    end

    def start!
      @stop_requested = false
      until stop_requested?
        loop
        sleep 1 unless stop_requested?
      end
    end

    def stop!
      @stop_requested = true
      puts "stop requested"
      EM.stop_event_loop if EM.reactor_running?
    end

    def restart!
      stop!
      start!
    end

    def loop
      EventMachine.run do
        Signal.trap("INT")  { stop! }
        Signal.trap("TERM") { stop! }
      end
    end

    def stop_requested?
      @stop_requested
    end

    def request(name, kind, message, options = {timeout: 0.1}, &block)
      options[:routing_key] = name
      queue_service.call(kind, message, options, &block)
    end

    def respond_to(service_instance, handler, options)
      EventMachine.next_tick do
        queue_options = {}
        subscription_options = {}
        if options[:fast] == true
          queue_options[:auto_delete] = false
        else
          subscription_options[:ack] = true
        end
        queue_service.queue(handler.to_s, queue_options).subscribe(subscription_options) do |delivery_info, payload|
          respond service_instance, payload, delivery_info<commit_msg>support for "sync" calls<commit_after>module Combi
  class Queue < Bus

    def post_initialize
      unless @options[:init_queue] == false
        require 'queue_service'
        EventMachine.next_tick do
          QueueService.start(ConfigProvider.for(:amqp), rpc: :enabled)
        end
      end
    end

    def start!
      @stop_requested = false
      until stop_requested?
        loop
        sleep 1 unless stop_requested?
      end
    end

    def stop!
      @stop_requested = true
      puts "stop requested"
      EM.stop_event_loop if EM.reactor_running?
    end

    def restart!
      stop!
      start!
    end

    def loop
      EventMachine.run do
        Signal.trap("INT")  { stop! }
        Signal.trap("TERM") { stop! }
      end
    end

    def stop_requested?
      @stop_requested
    end

    def request(name, kind, message, options = {timeout: 0.1}, &block)
      options[:routing_key] = name
      if block.nil? || options[:async] == false
        queue_service.call(kind, message, options, &block)
      else
        Thread.new do
          begin
            queue_service.call(kind, message, options, &block)
          rescue => e
            puts e.message
            puts e.backtrace
            retry
          end
        end
      end
    end

    def respond_to(service_instance, handler, options)
      EventMachine.next_tick do
        queue_options = {}
        subscription_options = {}
        if options[:fast] == true
          queue_options[:auto_delete] = false
        else
          subscription_options[:ack] = true
        end
        queue_service.queue(handler.to_s, queue_options).subscribe(subscription_options) do |delivery_info, payload|
          respond service_instance, payload, delivery_info
{'fraction_non_alphanumeric': 0.08066023045780131, 'fraction_numerical': 0.0018685767673621925, 'mean_word_length': 2.0944123314065513, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 4, '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': '2849479', 'n_tokens_mistral': 939, 'n_tokens_neox': 926, 'n_words': 232}