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:
            # 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]
This method cancels a consumer. This does not affect already
dataset/ETHPy150Open pika/pika/pika/adapters/blocking_connection.py/BlockingChannel.basic_cancel