'This method cancels a consumer. This does not affect already
delivered messages, but it does mean the server will not send any more
messages for that consumer. The client may receive an arbitrary number
of messages in between sending the cancel method and receiving the
cancel-ok reply.
NOTE: When cancelling a no_ack=False consumer, this implementation
automatically Nacks and suppresses any incoming messages that have not
yet been dispatched to the consumer\'s callback. However, when cancelling
a no_ack=True consumer, this method will return any pending messages
that arrived before broker confirmed the cancellation.
:param str consumer_tag: Identifier for the consumer; the result of
passing a consumer_tag that was created on another channel is
undefined (bad things will happen)
:returns: (NEW IN pika 0.10.0) empty sequence for a no_ack=False
consumer; for a no_ack=True consumer, returns a (possibly empty)
sequence of pending messages that arrived before broker confirmed
the cancellation (this is done instead of via consumer\'s callback in
order to prevent reentrancy/recursion. Each message is four-tuple:
(channel, method, properties, body)
channel: BlockingChannel
method: spec.Basic.Deliver
properties: spec.BasicProperties
body: str or unicode'

def basic_cancel(self, consumer_tag):

    try:
        consumer_info = self._consumer_infos[consumer_tag]
    except KeyError:
        LOGGER.warn('User    is    attempting    to    cancel    an    unknown    consumer=%s;    already    cancelled    by    user    or    broker?', consumer_tag)
        return []
    try:
        assert (consumer_info.active or consumer_info.cancelled_by_broker), consumer_info.state
        assert (consumer_info.cancelled_by_broker or (consumer_tag in self._impl._consumers)), consumer_tag
        no_ack = consumer_info.no_ack
        consumer_info.state = _ConsumerInfo.TEARING_DOWN
        with _CallbackResult() as cancel_ok_result:
            if (not no_ack):
                pending_messages = self._remove_pending_deliveries(consumer_tag)
                if pending_messages:
                    for message in pending_messages:
                        self._impl.basic_reject(message.method.delivery_tag, requeue=True)
            self._impl.basic_cancel(callback=cancel_ok_result.signal_once, consumer_tag=consumer_tag, nowait=False)
            self._flush_output(cancel_ok_result.is_ready, (lambda : (consumer_tag not in self._impl._consumers)))
        if no_ack:
            return [(evt.method, evt.properties, evt.body) for evt in self._remove_pending_deliveries(consumer_tag)]
        else:
            messages = self._remove_pending_deliveries(consumer_tag)
            assert (not messages), messages
            return []
    finally:
        if (consumer_tag in self._consumer_infos):
            del self._consumer_infos[consumer_tag]
