lib/messaging/client.rb
def open_connection(uri, delay = nil)
      delay ||= config.reconnect_delay

      options = AMQP::Client.parse_connection_uri(uri)
      options.merge!(:heartbeat => config.heartbeat)

      res = AMQP.connect(options) do |connection, open_ok|
        # Handle TCP connection errors
        connection.on_tcp_connection_loss do |conn, settings|
          log.error("Connection to #{uri.inspect} lost, reconnecting")

          conn.periodically_reconnect(delay)
        end

        # Handle general errors
        connection.on_error do |conn, error|
          log.error("Connection to #{uri.inspect} lost, reconnecting")

          if (402..540).include?(error.reply_code)
            raise(MessagingError, "Connection exception: #{error.reply_text.inspect}")
          end

          conn.periodically_reconnect(delay)
        end

        log.debug("Connection to #{uri.inspect} started")
      end

      register_connection(res)
    end
def open_connection(uri, delay = nil)
      delay ||= config.reconnect_delay

      options = AMQP::Client.parse_connection_uri(uri)
      options.merge!(:heartbeat => config.heartbeat)

      res = AMQP.connect(options) do |connection, open_ok|
        # Handle TCP connection errors
        connection.on_tcp_connection_loss do |conn, settings|
          log.error("Connection to #{uri.inspect} lost, reconnecting")

          conn.periodically_reconnect(delay)
        end

        # Handle general errors
        connection.on_error do |conn, error|
          log.error("Connection to #{uri.inspect} lost, reconnecting")

          if (402..540).include?(error.reply_code)
            raise(MessagingError, "Connection exception: #{error.reply_text.inspect}")
          end

          conn.periodically_reconnect(delay)
        end

        log.debug("Connection to #{uri.inspect} started")
      end

      register_connection(res)
    end
def open_connection ( uri , delay = nil ) delay ||= config . reconnect_delay options = AMQP :: Client . parse_connection_uri ( uri ) options . merge! ( :heartbeat => config . heartbeat ) res = AMQP . connect ( options ) do | connection , open_ok | # Handle TCP connection errors connection . on_tcp_connection_loss do | conn , settings | log . error ( "Connection to #{uri.inspect} lost, reconnecting" ) conn . periodically_reconnect ( delay ) end # Handle general errors connection . on_error do | conn , error | log . error ( "Connection to #{uri.inspect} lost, reconnecting" ) if ( 402 .. 540 ) . include? ( error . reply_code ) raise ( MessagingError , "Connection exception: #{error.reply_text.inspect}" ) end conn . periodically_reconnect ( delay ) end log . debug ( "Connection to #{uri.inspect} started" ) end register_connection ( res ) end
Create an AMQP::Connection with auto-reconnect and error handling.

 @param uri [String] The AMQP URI to connect to.
 @param delay [Integer, nil] Time to delay between reconnection attempts.
 @return [AMQP::Connection]
 @api public
Create an AMQP :: Connection with auto - reconnect and error handling .