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