code: def _read(self, queue_object, consumer):
          
          while self._running:
              try:
                  _channel, delivery_frame, properties, body = yield queue_object.get()
              except (error.ConnectionDone, ChannelClosedByClient):
                  # This is deliberate.
                  _legacy_twisted_log.msg(
                      "Stopping the AMQP consumer with tag {tag}", tag=consumer.tag
                  )
                  break
              except pika.exceptions.ChannelClosed as e:
                  _legacy_twisted_log.msg(
                      "Stopping AMQP consumer {tag} for queue {q}: {e}",
                      tag=consumer.tag,
                      q=consumer.queue,
                      e=str(e),
                      logLevel=logging.ERROR,
                  )
                  break
              except pika.exceptions.ConsumerCancelled as e:
                  _legacy_twisted_log.msg(
                      "The AMQP broker canceled consumer {tag} on queue {q}: {e}",
                      tag=consumer.tag,
                      q=consumer.queue,
                      e=str(e),
                      logLevel=logging.ERROR,
                  )
                  break
              except Exception:
                  _legacy_twisted_log.msg(
                      "An unexpected error occurred consuming from queue {q}",
                      q=consumer.queue,
                      logLevel=logging.ERROR,
                  )
                  break
              if body:
                  yield self._on_message(delivery_frame, properties, body, consumer)
docs: The loop that reads from the message queue and calls the consumer callback
          wrapper.
  
          queue_object (pika.adapters.twisted_connection.ClosableDeferredQueue):
              The AMQP queue the consumer is bound to.
          consumer (dict): A dictionary describing the consumer for the given
              queue_object.