Method close

pika/channel.py:576–616

Invoke a graceful shutdown of the channel with the AMQP Broker. If channel is OPENING, transition to CLOSING and suppress the incoming Channel.OpenOk, if any. :param int reply_code: The reason code to send to broker :param str reply_text: The reason text to send to

(self,
              reply_code: int = 0,
              reply_text: str = "Normal shutdown")

                         [spec.Basic.RecoverOk])

    def close(self,
              reply_code: int = 0,
              reply_text: str = "Normal shutdown") -> None:
        """Invoke a graceful shutdown of the channel with the AMQP Broker.

        If channel is OPENING, transition to CLOSING and suppress the incoming
        Channel.OpenOk, if any.

        :param int reply_code: The reason code to send to broker
        :param str reply_text: The reason text to send to broker

        :raises ChannelWrongStateError: if channel is closed or closing

        """
        if self.is_closed or self.is_closing:
            # Whoever is calling `close` might expect the on-channel-close-cb
            # to be called, which won't happen when it's already closed.
            self._raise_if_not_open()

        # If channel is OPENING, we will transition it to CLOSING state,
        # causing the _on_openok method to suppress the OPEN state transition
        # and the on-channel-open-callback

        LOGGER.info('Closing channel (%s): %r on %s', reply_code, reply_text,
                    self)

        # Save the reason info so that we may use it in the '_on_channel_close'
        # callback processing
        self._closing_reason = exceptions.ChannelClosedByClient(
            reply_code, reply_text)

        for consumer_tag in list(self._consumers.keys()):
            if consumer_tag not in self._cancelled:
                self.basic_cancel(consumer_tag=consumer_tag)

        # Change state after cancelling consumers to avoid
        # ChannelWrongStateError exception from basic_cancel
        self._set_state(self.CLOSING)

        self._rpc(spec.Channel.Close(reply_code, reply_text, 0, 0),
                  self._on_closeok, [spec.Channel.CloseOk])

    def confirm_delivery(
            self,