<commit_before>gem "amqp", "1.3.0"

require "amqp"

require File.join(File.dirname(__FILE__), "base")

module Sensu
  module Transport
    class RabbitMQ < Base
      def initialize
        super
        @queues = {}
      end

      def connect(options={})
        set_connection_options(options)
        create_connection_timeout
        connect_with_eligible_options
      end

      def reconnect
        unless @connection.reconnecting?
          @connection_timeout.cancel
          @before_reconnect.call
          timer = EM::PeriodicTimer.new(5) do
            begin
              @connection.reconnect_to(next_connection_options)
            rescue EventMachine::ConnectionError
            end
          end
          @channel.on_recovery do
            timer.cancel
            @after_reconnect.call
          end
        end
      end

      def connected?
        @connection.connected?
      end

      def close
        callback = Proc.new { @connection.close }
        connected? ? callback.call : EM.next_tick(callback)
      end

      def publish(exchange_type, exchange_name, message, options={}, &callback)
        begin
          @channel.method(exchange_type.to_sym).call(exchange_name, options).publish(message) do
            info = {}
            callback.call(info) if callback
          end
        rescue => error
          info = {:error => error}
          callback.call(info) if callback
        end
      end

      def subscribe(exchange_type, exchange_name, queue_name="", options={}, &callback)
        previously_declared = @queues.has_key?(queue_name)
        @queues[queue_name] ||= @channel.queue!(queue_name, :auto_delete => true)
        queue = @queues[queue_name]
        queue.bind(@channel.method(exchange_type.to_sym).call(exchange_name))
        unless previously_declared
          queue.subscribe(options, &callback)
        end
      end

      def unsubscribe(&callback)
        @queues.values.each do |queue|
          if connected?
            queue.unsubscribe
          else
            queue.before_recovery do
              queue.unsubscribe
            end
          end
        end
        @queues = {}
        @channel.recover if connected?
        super
      end

      def acknowledge(info, &callback)
        info.ack
        callback.call(info) if callback
      end

      def stats(queue_name, options={}, &callback)
        options = options.merge(:auto_delete => true)
        @channel.queue(queue_name, options).status do |messages, consumers|
          info = {
            :messages => messages,
            :consumers => consumers
          }
          callback.call(info)
        end
      end

      private

      def set_connection_options(options)
        @connection_options = Array(options)
      end

      def create_connection_timeout
        @connection_timeout = EM::Timer.new(20) do
          error = Error.new("timed out while attempting to connect to rabbitmq")
          @on_error.call(error)<commit_msg>[multiple-brokers] reconnect on initial connection timeout, to guard against hanging connections<commit_after>gem "amqp", "1.3.0"

require "amqp"

require File.join(File.dirname(__FILE__), "base")

module Sensu
  module Transport
    class RabbitMQ < Base
      def initialize
        super
        @queues = {}
      end

      def connect(options={})
        set_connection_options(options)
        create_connection_timeout
        connect_with_eligible_options
      end

      def reconnect
        unless @connection.reconnecting?
          @connection_timeout.cancel
          @before_reconnect.call
          timer = EM::PeriodicTimer.new(5) do
            begin
              @connection.reconnect_to(next_connection_options)
            rescue EventMachine::ConnectionError
            end
          end
          @channel.on_recovery do
            timer.cancel
            @after_reconnect.call
          end
        end
      end

      def connected?
        @connection.connected?
      end

      def close
        callback = Proc.new { @connection.close }
        connected? ? callback.call : EM.next_tick(callback)
      end

      def publish(exchange_type, exchange_name, message, options={}, &callback)
        begin
          @channel.method(exchange_type.to_sym).call(exchange_name, options).publish(message) do
            info = {}
            callback.call(info) if callback
          end
        rescue => error
          info = {:error => error}
          callback.call(info) if callback
        end
      end

      def subscribe(exchange_type, exchange_name, queue_name="", options={}, &callback)
        previously_declared = @queues.has_key?(queue_name)
        @queues[queue_name] ||= @channel.queue!(queue_name, :auto_delete => true)
        queue = @queues[queue_name]
        queue.bind(@channel.method(exchange_type.to_sym).call(exchange_name))
        unless previously_declared
          queue.subscribe(options, &callback)
        end
      end

      def unsubscribe(&callback)
        @queues.values.each do |queue|
          if connected?
            queue.unsubscribe
          else
            queue.before_recovery do
              queue.unsubscribe
            end
          end
        end
        @queues = {}
        @channel.recover if connected?
        super
      end

      def acknowledge(info, &callback)
        info.ack
        callback.call(info) if callback
      end

      def stats(queue_name, options={}, &callback)
        options = options.merge(:auto_delete => true)
        @channel.queue(queue_name, options).status do |messages, consumers|
          info = {
            :messages => messages,
            :consumers => consumers
          }
          callback.call(info)
        end
      end

      private

      def set_connection_options(options)
        @connection_options = Array(options)
      end

      def create_connection_timeout
        @connection_timeout = EM::Timer.new(20) do
          reconnect
{'fraction_non_alphanumeric': 0.08394648829431438, 'fraction_numerical': 0.002006688963210702, 'mean_word_length': 1.9289911851126347, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, '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': '2245897', 'n_tokens_mistral': 1619, 'n_tokens_neox': 1573, 'n_words': 354}