Closes the WebSocket connection.
(self, code: Optional[int] = None, reason: Optional[str] = None)
| 1269 | return None |
| 1270 | |
| 1271 | def close(self, code: Optional[int] = None, reason: Optional[str] = None) -> None: |
| 1272 | """Closes the WebSocket connection.""" |
| 1273 | if not self.server_terminated: |
| 1274 | if not self.stream.closed(): |
| 1275 | if code is None and reason is not None: |
| 1276 | code = 1000 # "normal closure" status code |
| 1277 | if code is None: |
| 1278 | close_data = b"" |
| 1279 | else: |
| 1280 | close_data = struct.pack(">H", code) |
| 1281 | if reason is not None: |
| 1282 | close_data += utf8(reason) |
| 1283 | try: |
| 1284 | self._write_frame(True, 0x8, close_data) |
| 1285 | except StreamClosedError: |
| 1286 | self._abort() |
| 1287 | self.server_terminated = True |
| 1288 | if self.client_terminated: |
| 1289 | if self._waiting is not None: |
| 1290 | self.stream.io_loop.remove_timeout(self._waiting) |
| 1291 | self._waiting = None |
| 1292 | self.stream.close() |
| 1293 | elif self._waiting is None: |
| 1294 | # Give the client a few seconds to complete a clean shutdown, |
| 1295 | # otherwise just close the connection. |
| 1296 | self._waiting = self.stream.io_loop.add_timeout( |
| 1297 | self.stream.io_loop.time() + 5, self._abort |
| 1298 | ) |
| 1299 | if self._ping_coroutine: |
| 1300 | self._ping_coroutine.cancel() |
| 1301 | self._ping_coroutine = None |
| 1302 | |
| 1303 | def is_closing(self) -> bool: |
| 1304 | """Return ``True`` if this connection is closing. |
no test coverage detected