Method close

websocket/_core.py:503–554

Close Websocket object Parameters ---------- status: int Status code to send. See VALID_CLOSE_STATUS in ABNF. reason: bytes The reason to close in UTF-8. timeout: int or float Timeout until receive a close frame.

(self, status: int = STATUS_NORMAL, reason: bytes = b"", timeout: int = 3)

        self.send(struct.pack("!H", status) + reason, ABNF.OPCODE_CLOSE)

    def close(self, status: int = STATUS_NORMAL, reason: bytes = b"", timeout: int = 3):
        """
        Close Websocket object

        Parameters
        ----------
        status: int
            Status code to send. See VALID_CLOSE_STATUS in ABNF.
        reason: bytes
            The reason to close in UTF-8.
        timeout: int or float
            Timeout until receive a close frame.
            If None, it will wait forever until receive a close frame.
        """
        if not self.connected:
            return
        if status < 0 or status >= ABNF.LENGTH_16:
            raise ValueError("code is invalid range")

        try:
            self.connected = False
            self.send(struct.pack("!H", status) + reason, ABNF.OPCODE_CLOSE)
            if self.sock is None:
                return
            sock_timeout = self.sock.gettimeout()
            self.sock.settimeout(timeout)
            start_time = time.time()
            while timeout is None or time.time() - start_time < timeout:
                try:
                    frame = self.recv_frame()
                    if frame.opcode != ABNF.OPCODE_CLOSE:
                        continue
                    if isEnabledForError():
                        recv_status = struct.unpack("!H", frame.data[0:2])[0]
                        if recv_status >= 3000 and recv_status <= 4999:
                            debug(f"close status: {repr(recv_status)}")
                        elif recv_status != STATUS_NORMAL:
                            error(f"close status: {repr(recv_status)}")
                    break
                except (
                    WebSocketConnectionClosedException,
                    WebSocketTimeoutException,
                    struct.error,
                ):
                    break
            if self.sock is not None:
                self.sock.settimeout(sock_timeout)
                self.sock.shutdown(socket.SHUT_RDWR)
        except:
            pass

        self.shutdown()

    def abort(self):
        """