lib/ghtorrent/command.rb
def queue_client(queue, key = queue, ack = :after, block)

      stopped = false
      while not stopped
        begin
          conn = Bunny.new(:host => config(:amqp_host),
                           :port => config(:amqp_port),
                           :username => config(:amqp_username),
                           :password => config(:amqp_password))
          conn.start

          ch  = conn.create_channel
          debug "Queue setting prefetch to #{config(:amqp_prefetch)}"
          ch.prefetch(config(:amqp_prefetch))
          debug "Queue connection to #{config(:amqp_host)} succeeded"

          x = ch.topic(config(:amqp_exchange), :durable => true,
                       :auto_delete => false)
          q = ch.queue(queue, :durable => true)
          q.bind(x, :routing_key => key)

          q.subscribe(:block => true,
                      :manual_ack => true) do |delivery_info, properties, msg|

            if ack == :before
              ch.acknowledge(delivery_info.delivery_tag)
            end

            begin
              block.call(msg)
            ensure
              if ack != :before
                ch.acknowledge(delivery_info.delivery_tag)
              end
            end
          end

        rescue Bunny::TCPConnectionFailed => e
          warn "Connection to #{config(:amqp_host)} failed. Retrying in 1 sec"
          sleep(1)
        rescue Bunny::PossibleAuthenticationFailureError => e
          warn "Could not authenticate as #{conn.username}"
        rescue Bunny::NotFound, Bunny::AccessRefused, Bunny::PreconditionFailed => e
          warn "Channel error: #{e}. Retrying in 1 sec"
          sleep(1)
        rescue Interrupt => _
          stopped = true
        rescue StandardError => e
          raise e
        end
      end

      ch.close unless ch.nil?
      conn.close unless conn.nil?

    end
def queue_client(queue, key = queue, ack = :after, block)

      stopped = false
      while not stopped
        begin
          conn = Bunny.new(:host => config(:amqp_host),
                           :port => config(:amqp_port),
                           :username => config(:amqp_username),
                           :password => config(:amqp_password))
          conn.start

          ch  = conn.create_channel
          debug "Queue setting prefetch to #{config(:amqp_prefetch)}"
          ch.prefetch(config(:amqp_prefetch))
          debug "Queue connection to #{config(:amqp_host)} succeeded"

          x = ch.topic(config(:amqp_exchange), :durable => true,
                       :auto_delete => false)
          q = ch.queue(queue, :durable => true)
          q.bind(x, :routing_key => key)

          q.subscribe(:block => true,
                      :manual_ack => true) do |delivery_info, properties, msg|

            if ack == :before
              ch.acknowledge(delivery_info.delivery_tag)
            end

            begin
              block.call(msg)
            ensure
              if ack != :before
                ch.acknowledge(delivery_info.delivery_tag)
              end
            end
          end

        rescue Bunny::TCPConnectionFailed => e
          warn "Connection to #{config(:amqp_host)} failed. Retrying in 1 sec"
          sleep(1)
        rescue Bunny::PossibleAuthenticationFailureError => e
          warn "Could not authenticate as #{conn.username}"
        rescue Bunny::NotFound, Bunny::AccessRefused, Bunny::PreconditionFailed => e
          warn "Channel error: #{e}. Retrying in 1 sec"
          sleep(1)
        rescue Interrupt => _
          stopped = true
        rescue StandardError => e
          raise e
        end
      end

      ch.close unless ch.nil?
      conn.close unless conn.nil?

    end
Specify a handler to incoming messages from a connection to a queue.

 @param queue [String] the queue name to bind to
 @param key [String] routing key for msgs for binding the queue to the exchange.
 @param ack [Symbol] when should acks be send, :before or :after the block returns
 @param block [Block]: A block accepting one argument (the message)