| 313 | self._echo_message(message, f) |
| 314 | |
| 315 | def websocket_end(self, f: http.HTTPFlow): |
| 316 | assert f.websocket is not None # satisfy type checker |
| 317 | if self.match(f): |
| 318 | if f.websocket.close_code in {1000, 1001, 1005}: |
| 319 | c = "client" if f.websocket.closed_by_client else "server" |
| 320 | self.echo( |
| 321 | f"WebSocket connection closed by {c}: {f.websocket.close_code} {f.websocket.close_reason}" |
| 322 | ) |
| 323 | else: |
| 324 | error = flow.Error( |
| 325 | f"WebSocket Error: {self.format_websocket_error(f.websocket)}" |
| 326 | ) |
| 327 | self.echo( |
| 328 | f"Error in WebSocket connection to {human.format_address(f.server_conn.address)}: {error}", |
| 329 | fg="red", |
| 330 | ) |
| 331 | |
| 332 | def format_websocket_error(self, websocket: WebSocketData) -> str: |
| 333 | try: |