<commit_before>require './app/models/agent'


module AmqpManager

  class << self

    def rails_channel
      Thread.current[:rails_channel] ||= @connection.create_channel
    end

    def rails_xchange
      Thread.current[:rails_xchange] ||= rails_channel.topic('voice.rails', auto_delete: false)
    end


    def custom_channel
      Thread.current[:custom_channel] ||= @connection.create_channel
    end

    def custom_xchange
      Thread.current[:custom_xchange] ||= custom_channel.topic('voice.custom', auto_delete: false)
    end


    def numbers_channel
      Thread.current[:numbers_channel] ||= @connection.create_channel
    end

    def numbers_xchange
      Thread.current[:numbers_xchange] ||= numbers_channel.topic('voice.numbers', auto_delete: false)
    end


    def publish(payload, include_custom, include_numbers)
      data = Marshal.dump(payload)

      rails_xchange.publish(data,   routing_key: 'voice.rails')
      custom_xchange.publish(data,  routing_key: 'voice.custom')  if include_custom
      numbers_xchange.publish(data, routing_key: 'voice.numbers') if include_numbers
    end


    def ahn_channel
      Thread.current[:ahn_channel] ||= @connection.create_channel
    end

    def ahn_xchange
      Thread.current[:ahn_xchange] ||= ahn_channel.topic('voice.ahn', auto_delete: false)
    end

    def ahn_queue
      Thread.current[:ahn_queue] ||= ahn_channel.queue('voice.ahn', auto_delete: false)
    end


    def shutdown!
      @connection.close
    end


    def establish_connection
      @connection = Bunny.new(
        host:     AhnConfig.rabbit_host,
        user:     AhnConfig.rabbit_user,
        password: AhnConfig.rabbit_pass
      ).tap { |c| c.start }
    rescue Bunny::TCPConnectionFailed
      sleep 1
      retry
    end


    def start
      establish_connection
      ahn_queue.bind(ahn_xchange, routing_key: 'voice.ahn')

      ahn_queue.subscribe { |delivery_info, metadata, payload|
        data = Marshal.load(payload)

        if data[:user_id]
          Agent.update_client_settings_with(data)
        elsif data[:command]
          Call.execute_command_with(data)
        end
      }
    end<commit_msg>close amqp channels after threaded sending<commit_after>require './app/models/agent'


module AmqpManager
  TOPICS = [:rails, :numbers, :custom, :ahn]

  class << self

    def close_channels
      TOPICS.each { |name| Thread.current["#{name}_channel".to_sym].try(:close) }
    end


    TOPICS.each { |name|
      define_method "#{name}_channel" do
        Thread.current["#{name}_channel".to_sym] ||= connection.create_channel
      end

      define_method "#{name}_xchange" do
        Thread.current["#{name}_xchange".to_sym] ||= send("#{name}_channel").topic("voice.#{name}", auto_delete: false)
      end

      define_method "#{name}_queue" do
        Thread.current["#{name}_queue".to_sym] ||= send("#{name}_channel").queue("voice.#{name}", auto_delete: false)
      end
    }


    def publish(payload, include_custom, include_numbers)
      Thread.new {
        data = Marshal.dump(payload)

        rails_xchange.publish(data,   routing_key: 'voice.rails')
        custom_xchange.publish(data,  routing_key: 'voice.custom')  if include_custom
        numbers_xchange.publish(data, routing_key: 'voice.numbers') if include_numbers

        close_channels
      }
    end


    def connection
      establish_connection unless @connection
      @connection
    end


    def shutdown!
      connection.close
    end


    def establish_connection
      @connection = Bunny.new(
        host:     AhnConfig.rabbit_host,
        user:     AhnConfig.rabbit_user,
        password: Ahn<PASSWORD>
      ).tap { |c| c.start }
    rescue Bunny::TCPConnectionFailed
      sleep 1
      retry
    end


    def start
      establish_connection
      ahn_queue.bind(ahn_xchange, routing_key: 'voice.ahn')

      ahn_queue.subscribe { |delivery_info, metadata, payload|
        data = Marshal.load(payload)

        if data[:user_id]
          Agent.update_client_settings_with(data)
        elsif data[:command]
          Call.execute_command_with(data)
        end
      }
    end
{'fraction_non_alphanumeric': 0.10423688011555128, 'fraction_numerical': 0.00048146364949446316, 'mean_word_length': 2.8579387186629526, 'pattern_counts': {'":': 0, '<': 8, '<?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': '614750', 'n_tokens_mistral': 1319, 'n_tokens_neox': 1238, 'n_words': 241}