lib/messaging/consumer.rb
def subscribe(exchange, type, queue, key)
      consumer_channels.each do |channel|
        ex = declare_exchange(channel, exchange, type, config.exchange_options)
        q  = declare_queue(channel, ex, queue, key, config.queue_options)

        # Expliclity create an AMQP::Consumer rather than using
        # AMQP::Queue.subscription, which is a global singleton
        # and prevents the creation of multiple subscriptions
        AMQP::Consumer.new(channel, q).consume.on_delivery do |meta, payload|
          log.debug("Receieved message on channel #{meta.channel.id} from queue #{queue.inspect}")

          # If an exception is raised in on_message, the message will not be
          # acknowledged and the exception will be logged and re-raised
          begin
            on_message(meta, payload)

            meta.ack
          rescue => ex
            log.error("Exception: #{ex}, " \
              "Payload: #{payload.inspect}, " \
              "Headers: #{meta.headers.inspect}\n" \
              "Backtrace:\n#{ex.backtrace.join('\n')}")

            # Re-raise the exception
            raise ex
          end
        end
      end

      self
    end
def subscribe(exchange, type, queue, key)
      consumer_channels.each do |channel|
        ex = declare_exchange(channel, exchange, type, config.exchange_options)
        q  = declare_queue(channel, ex, queue, key, config.queue_options)

        # Expliclity create an AMQP::Consumer rather than using
        # AMQP::Queue.subscription, which is a global singleton
        # and prevents the creation of multiple subscriptions
        AMQP::Consumer.new(channel, q).consume.on_delivery do |meta, payload|
          log.debug("Receieved message on channel #{meta.channel.id} from queue #{queue.inspect}")

          # If an exception is raised in on_message, the message will not be
          # acknowledged and the exception will be logged and re-raised
          begin
            on_message(meta, payload)

            meta.ack
          rescue => ex
            log.error("Exception: #{ex}, " \
              "Payload: #{payload.inspect}, " \
              "Headers: #{meta.headers.inspect}\n" \
              "Backtrace:\n#{ex.backtrace.join('\n')}")

            # Re-raise the exception
            raise ex
          end
        end
      end

      self
    end
Subscribe to a queue which will invoke the supplied block when
 a message is received.
 Additionally declaring a binding to the specified exchange/key pair.

 Evaluation: Eager - this will be evaluated when called.
 Calls to #consume are not required.

 @param exchange [String]
 @param type [String]
 @param queue [String]
 @param key [String]
 @return [Messaging::Consumer]
 @api public