'Blocking consumption of a queue instead of via a callback. This
method is a generator that yields each message as a tuple of method,
properties, and body. The active generator iterator terminates when the
consumer is cancelled by client or broker.
Example:
for method, properties, body in channel.consume(\'queue\'):
print body
channel.basic_ack(method.delivery_tag)
You should call `BlockingChannel.cancel()` when you escape out of the
generator loop.
If you don\'t cancel this consumer, then next call on the same channel
to `consume()` with the exact same (queue, no_ack, exclusive) parameters
will resume the existing consumer generator; however, calling with
different parameters will result in an exception.
:param queue: The queue name to consume
:type queue: str or unicode
:param bool no_ack: Tell the broker to not expect a ack/nack response
:param bool exclusive: Don\'t allow other consumers on the queue
:param dict arguments: Custom key/value pair arguments for the consumer
:param float inactivity_timeout: if a number is given (in
seconds), will cause the method to yield None after the given period
of inactivity; this permits for pseudo-regular maintenance
activities to be carried out by the user while waiting for messages
to arrive. If None is given (default), then the method blocks until
the next event arrives. NOTE that timing granularity is limited by
the timer resolution of the underlying implementation.
NEW in pika 0.10.0.
:yields: tuple(spec.Basic.Deliver, spec.BasicProperties, str or unicode)
:raises ValueError: if consumer-creation parameters don\'t match those
of the existing queue consumer generator, if any.
NEW in pika 0.10.0'

def consume(self, queue, no_ack=False, exclusive=False, arguments=None, inactivity_timeout=None):

    params = (queue, no_ack, exclusive)
    if (self._queue_consumer_generator is not None):
        if (params != self._queue_consumer_generator.params):
            raise ValueError(('Consume    with    different    params    not    allowed    on    existing    queue    consumer    generator;    previous    params:    %r;    new    params:    %r' % (self._queue_consumer_generator.params, (queue, no_ack, exclusive))))
    else:
        LOGGER.debug('Creating    new    queue    consumer    generator;    params:    %r', params)
        consumer_tag = self._impl._generate_consumer_tag()
        self._queue_consumer_generator = _QueueConsumerGeneratorInfo(params, consumer_tag)
        try:
            self._basic_consume_impl(queue=queue, no_ack=no_ack, exclusive=exclusive, consumer_tag=consumer_tag, arguments=arguments, alternate_event_sink=self._on_consumer_generator_event)
        except Exception:
            self._queue_consumer_generator = None
            raise
        LOGGER.info('Created    new    queue    consumer    generator    %r', self._queue_consumer_generator)
    while (self._queue_consumer_generator is not None):
        if self._queue_consumer_generator.pending_events:
            evt = self._queue_consumer_generator.pending_events.popleft()
            if (type(evt) is _ConsumerCancellationEvt):
                self._queue_consumer_generator = None
                break
            else:
                (yield (evt.method, evt.properties, evt.body))
                continue
        if (inactivity_timeout is None):
            self.connection.process_data_events(time_limit=None)
            continue
        wait_start_time = time.time()
        wait_deadline = (wait_start_time + inactivity_timeout)
        delta = inactivity_timeout
        while ((self._queue_consumer_generator is not None) and (not self._queue_consumer_generator.pending_events)):
            self.connection.process_data_events(time_limit=delta)
            if (not self._queue_consumer_generator):
                break
            if self._queue_consumer_generator.pending_events:
                break
            delta = (wait_deadline - time.time())
            if (delta <= 0.0):
                (yield None)
                break
