<commit_before>gem "amqp", "1.5.0"

require "amqp"

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

module Sensu
  module Transport
    class RabbitMQ < Base
      # RabbitMQ connection setup.
      #
      # @param options [Hash, String]
      def connect(options={})
        reset
        set_connection_options(options)
        create_connection_timeout
        connect_with_eligible_options
      end

      # Reconnect to RabbitMQ.
      #
      # @param force [Boolean] the reconnect.
      def reconnect(force=false)
        unless @reconnecting
          @reconnecting = true
          @before_reconnect.call
          reset
          periodically_reconnect
        end
      end

      # Indicates if connected to RabbitMQ.
      #
      # @return [TrueClass, FalseClass]
      def connected?
        @connection.connected?
      end

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

      # Publish a message to RabbitMQ.
      #
      # @param type [Symbol] the RabbitMQ exchange type, possible
      #   values are: :direct and :fanout.
      # @param pipe [String] the RabbitMQ exchange name.
      # @param message [String] the message to be published to
      #   RabbitMQ.
      # @param options [Hash] the options to publish the message with.
      # @yield [info] passes publish info to an optional
      #   callback/block.
      # @yieldparam info [Hash] contains publish information.
      def publish(type, pipe, message, options={}, &callback)
        catch_errors do
          @channel.method(type.to_sym).call(pipe, options).publish(message) do
            info = {}
            callback.call(info) if callback
          end
        end
      end

      # Subscribe to a RabbitMQ queue.
      #
      # @param type [Symbol] the RabbitMQ exchange type, possible
      #   values are: :direct and :fanout.
      # @param pipe [String] the RabbitMQ exhange name.
      # @param funnel [String] the RabbitMQ queue.
      # @param options [Hash] the options to consume messages with.
      # @yield [info, message] passes message info and content to the
      #   consumer callback/block.
      # @yieldparam info [Hash] contains message information.
      # @yieldparam message [String] message.
      def subscribe(type, pipe, funnel="", options={}, &callback)
        catch_errors do
          previously_declared = @queues.has_key?(funnel)
          @queues[funnel] ||= @channel.queue!(funnel, :auto_delete => true)
          queue = @queues[funnel]
          queue.bind(@channel.method(type.to_sym).call(pipe))
          unless previously_declared
            queue.subscribe(options, &callback)
          end
        end
      end

      # Unsubscribe from all RabbitMQ queues.
      #
      # @yield [info] passes info to an optional callback/block.
      # @yieldparam info [Hash] contains unsubscribe information.
      def unsubscribe(&callback)
        catch_errors do
          @queues.values.each do |queue|
            if connected?
              queue.unsubscribe
            else
              queue.before_recovery do
                queue.unsubscribe
              end
            end
          end
          @queues = {}
          @channel.recover if connected?
        end
        super
      end

      # Acknowledge the delivery of a message from RabbitMQ.
      #
      # @param info [Hash] message info containing its delivery tag.
      # @yield [info] passes acknowledgment info to an optional
      #   callback/block.
      def acknowledge(info, &callback)
        catch_errors do
          info.ack
        end
        callback.call(info) if callback
      end

      # RabbitMQ queue stats, including message and consumer counts.
      #
      # @param funnel [String] the RabbitMQ queue to get stats for.
      # @param options [Hash] the options to get queue stats with.
      # @yield [info] passes queue stats to the callback/block.
      def stats(funnel, options={}, &callback)
        catch_errors do
          options = options.merge(:auto_delete => true)
          @channel.queue(funnel, options).status do |messages, consumers|
            info = {
              :messages => messages,
              :consumers => consumers
            }
            callback.call(info)
          end
        end
      end

      private

      # Catch RabbitMQ errors and call the on_error callback,
      # providing it with the error object as an argument. This method
      # is intended to be applied where necessary, not to be confused
      # with a catch-all.
      #
      # @param block [Proc] called within a rescue block to
      #   catch RabbitMQ errors.
      def catch_errors(&block)
        begin
          block.call
        rescue AMQP => error
          @on_error.call(error)
        end
      end

      def reset
        @queues = {}
        @connection_timeout.cancel if @connection_timeout
        @connection.close_connection if @connection
      end

      def set_connection_options(options)
        @connection_options = [options].flatten
      end

      def create_connection_timeout
        @connection_timeout = EM::Timer.new(20) do
          reconnect
        end
      end

      def next_connection_options
        if @eligible_options.nil? || @eligible_options.empty?
          @eligible_options = @connection_options.shuffle
        end
        @eligible_options.shift
      end

      def reconnect_callback
        Proc.new { reconnect }
      end<commit_msg>use correct rabbitmq amqp error class<commit_after>gem "amqp", "1.5.0"

require "amqp"

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

module Sensu
  module Transport
    class RabbitMQ < Base
      # RabbitMQ connection setup.
      #
      # @param options [Hash, String]
      def connect(options={})
        reset
        set_connection_options(options)
        create_connection_timeout
        connect_with_eligible_options
      end

      # Reconnect to RabbitMQ.
      #
      # @param force [Boolean] the reconnect.
      def reconnect(force=false)
        unless @reconnecting
          @reconnecting = true
          @before_reconnect.call
          reset
          periodically_reconnect
        end
      end

      # Indicates if connected to RabbitMQ.
      #
      # @return [TrueClass, FalseClass]
      def connected?
        @connection.connected?
      end

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

      # Publish a message to RabbitMQ.
      #
      # @param type [Symbol] the RabbitMQ exchange type, possible
      #   values are: :direct and :fanout.
      # @param pipe [String] the RabbitMQ exchange name.
      # @param message [String] the message to be published to
      #   RabbitMQ.
      # @param options [Hash] the options to publish the message with.
      # @yield [info] passes publish info to an optional
      #   callback/block.
      # @yieldparam info [Hash] contains publish information.
      def publish(type, pipe, message, options={}, &callback)
        catch_errors do
          @channel.method(type.to_sym).call(pipe, options).publish(message) do
            info = {}
            callback.call(info) if callback
          end
        end
      end

      # Subscribe to a RabbitMQ queue.
      #
      # @param type [Symbol] the RabbitMQ exchange type, possible
      #   values are: :direct and :fanout.
      # @param pipe [String] the RabbitMQ exhange name.
      # @param funnel [String] the RabbitMQ queue.
      # @param options [Hash] the options to consume messages with.
      # @yield [info, message] passes message info and content to the
      #   consumer callback/block.
      # @yieldparam info [Hash] contains message information.
      # @yieldparam message [String] message.
      def subscribe(type, pipe, funnel="", options={}, &callback)
        catch_errors do
          previously_declared = @queues.has_key?(funnel)
          @queues[funnel] ||= @channel.queue!(funnel, :auto_delete => true)
          queue = @queues[funnel]
          queue.bind(@channel.method(type.to_sym).call(pipe))
          unless previously_declared
            queue.subscribe(options, &callback)
          end
        end
      end

      # Unsubscribe from all RabbitMQ queues.
      #
      # @yield [info] passes info to an optional callback/block.
      # @yieldparam info [Hash] contains unsubscribe information.
      def unsubscribe(&callback)
        catch_errors do
          @queues.values.each do |queue|
            if connected?
              queue.unsubscribe
            else
              queue.before_recovery do
                queue.unsubscribe
              end
            end
          end
          @queues = {}
          @channel.recover if connected?
        end
        super
      end

      # Acknowledge the delivery of a message from RabbitMQ.
      #
      # @param info [Hash] message info containing its delivery tag.
      # @yield [info] passes acknowledgment info to an optional
      #   callback/block.
      def acknowledge(info, &callback)
        catch_errors do
          info.ack
        end
        callback.call(info) if callback
      end

      # RabbitMQ queue stats, including message and consumer counts.
      #
      # @param funnel [String] the RabbitMQ queue to get stats for.
      # @param options [Hash] the options to get queue stats with.
      # @yield [info] passes queue stats to the callback/block.
      def stats(funnel, options={}, &callback)
        catch_errors do
          options = options.merge(:auto_delete => true)
          @channel.queue(funnel, options).status do |messages, consumers|
            info = {
              :messages => messages,
              :consumers => consumers
            }
            callback.call(info)
          end
        end
      end

      private

      # Catch RabbitMQ errors and call the on_error callback,
      # providing it with the error object as an argument. This method
      # is intended to be applied where necessary, not to be confused
      # with a catch-all.
      #
      # @param block [Proc] called within a rescue block to
      #   catch RabbitMQ errors.
      def catch_errors(&block)
        begin
          block.call
        rescue AMQP::Error => error
          @on_error.call(error)
        end
      end

      def reset
        @queues = {}
        @connection_timeout.cancel if @connection_timeout
        @connection.close_connection if @connection
      end

      def set_connection_options(options)
        @connection_options = [options].flatten
      end

      def create_connection_timeout
        @connection_timeout = EM::Timer.new(20) do
          reconnect
        end
      end

      def next_connection_options
        if @eligible_options.nil? || @eligible_options.empty?
          @eligible_options = @connection_options.shuffle
        end
        @eligible_options.shift
      end

      def reconnect_callback
        Proc.new { reconnect }
      end
{'fraction_non_alphanumeric': 0.0782647189021691, 'fraction_numerical': 0.0008853474988933156, 'mean_word_length': 1.9820485744456178, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, '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': '948267', 'n_tokens_mistral': 3101, 'n_tokens_neox': 2953, 'n_words': 952}