def _on_connection_close(self, connection, reply_code_or_reason, reply_text=None):
        """
        Callback invoked when a previously-opened connection is closed.

        Args:
            connection (pika.connection.SelectConnection): The connection that
                was just closed.
            reply_code_or_reason (int|Exception): The reason why the channel
                was closed. In older versions of pika, this is the AMQP code.
            reply_text (str): The human-readable reason the connection was
                closed (only in older versions of pika)
        """
        self._channel = None

        if isinstance(reply_code_or_reason, pika_errs.ConnectionClosed):
            reply_code = reply_code_or_reason.reply_code
            reply_text = reply_code_or_reason.reply_text
        elif isinstance(reply_code_or_reason, int):
            reply_code = reply_code_or_reason
        else:
            reply_code = 0
            reply_text = str(reply_code_or_reason)

        if reply_code == 200:
            # Normal shutdown, exit the consumer.
            _log.info("Server connection closed (%s), shutting down", reply_text)
            connection.ioloop.stop()
        else:
            _log.warning(
                "Connection to %s closed unexpectedly (%d): %s",
                connection.params.host,
                reply_code,
                reply_text,
            )
            self.call_later(1, self.reconnect)