def subscribe(queue, exchange = nil, options = {}, &blk)
      identities = []
      brokers = options.delete(:brokers)
      each(:usable, brokers) { |b| identities << b.identity if b.subscribe(queue, exchange, options, &blk) }
      logger.info("Could not subscribe to queue #{queue.inspect} on exchange #{exchange.inspect} " +
                  "on brokers #{each(:usable, brokers).inspect} when selected #{brokers.inspect} " +
                  "from usable #{usable.inspect}") if identities.empty?
      identities
    end
def subscribe(queue, exchange = nil, options = {}, &blk)
      identities = []
      brokers = options.delete(:brokers)
      each(:usable, brokers) { |b| identities << b.identity if b.subscribe(queue, exchange, options, &blk) }
      logger.info("Could not subscribe to queue #{queue.inspect} on exchange #{exchange.inspect} " +
                  "on brokers #{each(:usable, brokers).inspect} when selected #{brokers.inspect} " +
                  "from usable #{usable.inspect}") if identities.empty?
      identities
    end
Subscribe an AMQP queue to an AMQP exchange on all broker clients that are connected or still connecting
 Allow connecting here because subscribing may happen before all have confirmed connected
 Do not wait for confirmation from broker client 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 a warning

 === 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 client 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 client does not have create permission or
       knows the exchange already exists and wants to avoid declare overhead
 options(Hash):: Subscribe options:
   :ack(Boolean):: Whether client 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

 === Block
 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)

 === Return
 identities(Array):: Identity of brokers where successfully subscribed