def basic_cancel(self, consumer_tag):
        """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
        """
        try:
            consumer_info = self._consumer_infos[consumer_tag]
        except __HOLE__:
            LOGGER.warn("User is attempting to cancel an unknown consumer=%s; "
                        "already cancelled by user or broker?", consumer_tag)
            return []

        try:
            # Assertion failure here is most likely due to reentrance
            assert consumer_info.active or consumer_info.cancelled_by_broker, (
                consumer_info.state)

            # Assertion failure here signals disconnect between consumer state
            # in BlockingChannel and Channel
            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:
                # Nack pending messages for no_ack=False consumer
                if not no_ack:
                    pending_messages = self._remove_pending_deliveries(
                        consumer_tag)
                    if pending_messages:
                        # NOTE: we use impl's basic_reject to avoid the
                        # possibility of redelivery before basic_cancel takes
                        # control of nacking.
                        # NOTE: we can't use basic_nack with the multiple option
                        # to avoid nacking messages already held by our client.
                        for message in pending_messages:
                            self._impl.basic_reject(message.method.delivery_tag,
                                                    requeue=True)

                # Cancel the consumer; impl takes care of rejecting any
                # additional deliveries that arrive for a no_ack=False
                # consumer
                self._impl.basic_cancel(
                    callback=cancel_ok_result.signal_once,
                    consumer_tag=consumer_tag,
                    nowait=False)

                # Flush output and wait for Basic.Cancel-ok or
                # broker-initiated Basic.Cancel
                self._flush_output(
                    cancel_ok_result.is_ready,
                    lambda: consumer_tag not in self._impl._consumers)

            if no_ack:
                # Return pending messages for no_ack=True consumer
                return [
                    (evt.method, evt.properties, evt.body)
                    for evt in self._remove_pending_deliveries(consumer_tag)]
            else:
                # impl takes care of rejecting any incoming deliveries during
                # cancellation
                messages = self._remove_pending_deliveries(consumer_tag)
                assert not messages, messages

                return []
        finally:
            # NOTE: The entry could be purged if channel or connection closes
            if consumer_tag in self._consumer_infos:
                del self._consumer_infos[consumer_tag]
dataset/ETHPy150Open pika/pika/pika/adapters/blocking_connection.py/BlockingChannel.basic_cancel