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)

        
        
        
        AMQP::Consumer.new(channel, q).consume.on_delivery do |meta, payload|
          log.debug("Receieved message on channel 

          
          
          begin
            on_message(meta, payload)

            meta.ack
          rescue => ex
            log.error("Exception: 
              "Payload: 
              "Headers: 
              "Backtrace:\n

            
            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
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 .