def subscribe(queue, exchange = nil, options = {}, &block)
      raise ArgumentError, "Must call this method with a block" unless block
      return false unless usable?
      return true unless @queues.select { |q| q.name == queue[:name] }.empty?

      to_exchange =  if exchange
        if options[:exchange2]
          " to exchanges #{exchange[:name]} and #{options[:exchange2][:name]}"
        else
          " to exchange #{exchange[:name]}"
        end
      end
      queue_options = queue[:options] || {}
      exchange_options = (exchange && exchange[:options]) || {}

      begin
        logger.info("[setup] Subscribing queue #{queue[:name]}#{to_exchange} on broker #{@alias}")
        q = @channel.queue(queue[:name], queue_options)
        @queues << q
        if exchange
          x = @channel.__send__(exchange[:type], exchange[:name], exchange_options)
          binding = q.bind(x, options[:key] ? {:key => options[:key]} : {})
          if (exchange2 = options[:exchange2])
            q.bind(@channel.__send__(exchange2[:type], exchange2[:name], exchange2[:options] || {}))
          end
          q = binding
        end
        q.subscribe(options[:ack] ? {:ack => true} : {}) do |header, message|
          begin
            if (pool = (options[:fiber_pool] || @options[:fiber_pool]))
              pool.spawn { receive(queue[:name], header, message, options, &block) }
            else
              receive(queue[:name], header, message, options, &block)
            end
          rescue StandardError => e
            header.ack if options[:ack]
            logger.exception("Failed setting up to receive message from queue #{queue.inspect} " +
                             "on broker #{@alias}", e, :trace)
            @exception_stats.track("receive", e)
            update_non_delivery_stats("receive failure", e)
          end
        end
      rescue StandardError => e
        logger.exception("Failed subscribing queue #{queue.inspect}#{to_exchange} on broker #{@alias}", e, :trace)
        @exception_stats.track("subscribe", e)
        false
      end
    end
def subscribe(queue, exchange = nil, options = {}, &block)
      raise ArgumentError, "Must call this method with a block" unless block
      return false unless usable?
      return true unless @queues.select { |q| q.name == queue[:name] }.empty?

      to_exchange =  if exchange
        if options[:exchange2]
          " to exchanges #{exchange[:name]} and #{options[:exchange2][:name]}"
        else
          " to exchange #{exchange[:name]}"
        end
      end
      queue_options = queue[:options] || {}
      exchange_options = (exchange && exchange[:options]) || {}

      begin
        logger.info("[setup] Subscribing queue #{queue[:name]}#{to_exchange} on broker #{@alias}")
        q = @channel.queue(queue[:name], queue_options)
        @queues << q
        if exchange
          x = @channel.__send__(exchange[:type], exchange[:name], exchange_options)
          binding = q.bind(x, options[:key] ? {:key => options[:key]} : {})
          if (exchange2 = options[:exchange2])
            q.bind(@channel.__send__(exchange2[:type], exchange2[:name], exchange2[:options] || {}))
          end
          q = binding
        end
        q.subscribe(options[:ack] ? {:ack => true} : {}) do |header, message|
          begin
            if (pool = (options[:fiber_pool] || @options[:fiber_pool]))
              pool.spawn { receive(queue[:name], header, message, options, &block) }
            else
              receive(queue[:name], header, message, options, &block)
            end
          rescue StandardError => e
            header.ack if options[:ack]
            logger.exception("Failed setting up to receive message from queue #{queue.inspect} " +
                             "on broker #{@alias}", e, :trace)
            @exception_stats.track("receive", e)
            update_non_delivery_stats("receive failure", e)
          end
        end
      rescue StandardError => e
        logger.exception("Failed subscribing queue #{queue.inspect}#{to_exchange} on broker #{@alias}", e, :trace)
        @exception_stats.track("subscribe", e)
        false
      end
    end
def subscribe ( queue , exchange = nil , options = { } , & block ) raise ArgumentError , "Must call this method with a block" unless block return false unless usable? return true unless @queues . select { | q | q . name == queue [ :name ] } . empty? to_exchange = if exchange if options [ :exchange2 ] " to exchanges #{exchange[:name]} and #{options[:exchange2][:name]}" else " to exchange #{exchange[:name]}" end end queue_options = queue [ :options ] || { } exchange_options = ( exchange && exchange [ :options ] ) || { } begin logger . info ( "[setup] Subscribing queue #{queue[:name]}#{to_exchange} on broker #{@alias}" ) q = @channel . queue ( queue [ :name ] , queue_options ) @queues << q if exchange x = @channel . __send__ ( exchange [ :type ] , exchange [ :name ] , exchange_options ) binding = q . bind ( x , options [ :key ] ? { :key => options [ :key ] } : { } ) if ( exchange2 = options [ :exchange2 ] ) q . bind ( @channel . __send__ ( exchange2 [ :type ] , exchange2 [ :name ] , exchange2 [ :options ] || { } ) ) end q = binding end q . subscribe ( options [ :ack ] ? { :ack => true } : { } ) do | header , message | begin if ( pool = ( options [ :fiber_pool ] || @options [ :fiber_pool ] ) ) pool . spawn { receive ( queue [ :name ] , header , message , options , block ) } else receive ( queue [ :name ] , header , message , options , block ) end rescue StandardError => e header . ack if options [ :ack ] logger . exception ( "Failed setting up to receive message from queue #{queue.inspect} " + "on broker #{@alias}" , e , :trace ) @exception_stats . track ( "receive" , e ) update_non_delivery_stats ( "receive failure" , e ) end end rescue StandardError => e logger . exception ( "Failed subscribing queue #{queue.inspect}#{to_exchange} on broker #{@alias}" , e , :trace ) @exception_stats . track ( "subscribe" , e ) false end end
Subscribe an AMQP queue to an AMQP exchange
 Do not wait for confirmation from broker that subscription is complete
 When a message is received, acknowledge, unserialize, and log it as specified
 If the message is unserialized and it is not of the right type, it is dropped after logging an error

 === Parameters
 queue(Hash):: AMQP queue being subscribed with keys :name and :options,
   which are the standard AMQP ones plus
     :no_declare(Boolean):: Whether to skip declaring this queue on the broker
       to cause its creation; for use when caller does not have permission to create or
       knows the queue already exists and wants to avoid declare overhead
 exchange(Hash|nil):: AMQP exchange to subscribe to with keys :type, :name, and :options,
   nil means use empty exchange by directly subscribing to queue; the :options are the
   standard AMQP ones plus
     :no_declare(Boolean):: Whether to skip declaring this exchange on the broker
       to cause its creation; for use when caller does not have create permission or
       knows the exchange already exists and wants to avoid declare overhead
 options(Hash):: Subscribe options:
   :ack(Boolean):: Whether caller takes responsibility for explicitly acknowledging each
     message received, defaults to implicit acknowledgement in AMQP as part of message receipt
   :no_unserialize(Boolean):: Do not unserialize message, this is an escape for special
     situations like enrollment, also implicitly disables receive filtering and logging;
     this option is implicitly invoked if initialize without a serializer
   (packet class)(Array(Symbol)):: Filters to be applied in to_s when logging packet to :info,
     only packet classes specified are accepted, others are not processed but are logged with error
   :category(String):: Packet category description to be used in error messages
   :log_data(String):: Additional data to display at end of log entry
   :no_log(Boolean):: Disable receive logging unless debug level
   :exchange2(Hash):: Additional exchange to which same queue is to be bound
   :brokers(Array):: Identity of brokers for which to subscribe, defaults to all usable if nil or empty
   :fiber_pool(FiberPool):: Pool of initialized fibers to be used for asynchronous message
     processing (non-nil value will override constructor option setting)

 === Block
 Required block with following parameters to be called each time exchange matches a message to the queue
   identity(String):: Serialized identity of broker delivering the message
   message(Packet|String):: Message received, which is unserialized unless :no_unserialize was specified
   header(AMQP::Protocol::Header):: Message header (optional block parameter)

 === Raise
 ArgumentError:: If a block is not supplied

 === Return
 (Boolean):: true if subscribe successfully or if already subscribed, otherwise false
Subscribe an AMQP queue to an AMQP exchange Do not wait for confirmation from broker that subscription is complete When a message is received acknowledge unserialize and log it as specified If the message is unserialized and it is not of the right type it is dropped after logging an error