Method basic_cancel

pika/channel.py:253–311

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

(self,
                     consumer_tag: str = '',
                     callback: Optional[_OnBasicCancelCallback] = None)

        return self._send_method(spec.Basic.Ack(delivery_tag, multiple))

    def basic_cancel(self,
                     consumer_tag: str = '',
                     callback: Optional[_OnBasicCancelCallback] = None) -> None:
        """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. It may also be sent from the server to the client in
        the event of the consumer being unexpectedly cancelled (i.e. cancelled
        for any reason other than the server receiving the corresponding
        basic.cancel from the client). This allows clients to be notified of
        the loss of consumers due to events such as queue deletion.

        :param str consumer_tag: Identifier for the consumer
        :param callable callback: callback(pika.frame.Method) for method
            Basic.CancelOk. If None, do not expect a Basic.CancelOk response,
            otherwise, callback must be callable

        :raises ValueError:

        """
        validators.require_string(consumer_tag, 'consumer_tag')
        self._raise_if_not_open()
        nowait = validators.rpc_completion_callback(callback)

        if consumer_tag in self._cancelled:
            # We check for cancelled first, because basic_cancel removes
            # consumers closed with nowait from self._consumers
            LOGGER.warning('basic_cancel - consumer is already cancelling: %s',
                           consumer_tag)
            return

        if consumer_tag not in self._consumers:
            # Could be cancelled by user or broker earlier
            LOGGER.warning('basic_cancel - consumer not found: %s',
                           consumer_tag)
            return

        LOGGER.debug('Cancelling consumer: %s (nowait=%s)', consumer_tag,
                     nowait)

        if nowait:
            # This is our last opportunity while the channel is open to remove
            # this consumer callback and help gc; unfortunately, this consumer's
            # self._cancelled and self._consumers_with_noack (if any) entries
            # will persist until the channel is closed.
            del self._consumers[consumer_tag]

        if callback is not None:
            self.callbacks.add(self.channel_number, spec.Basic.CancelOk,
                               callback)

        self._cancelled.add(consumer_tag)

        self._rpc(spec.Basic.Cancel(consumer_tag=consumer_tag, nowait=nowait),
                  self._on_cancelok if not nowait else None,
                  [(spec.Basic.CancelOk, {
                      'consumer_tag': consumer_tag