| 26 | |
| 27 | |
| 28 | class WebSocketConnection(ws_adaptor.WebSocketConnection): |
| 29 | |
| 30 | def __init__(self, websocket: WebSocket, ioloop): |
| 31 | self.ws = websocket |
| 32 | self.ioloop = ioloop |
| 33 | |
| 34 | def get_query_argument(self, name) -> typing.Optional[str]: |
| 35 | return self.ws.query_params.get(name, None) |
| 36 | |
| 37 | def make_session_info(self) -> dict: |
| 38 | session_info = get_session_info_from_headers(self.ws.headers) |
| 39 | session_info['user_ip'] = self.ws.client.host or '' |
| 40 | session_info['request'] = self.ws |
| 41 | session_info['backend'] = 'starlette' |
| 42 | session_info['protocol'] = 'websocket' |
| 43 | return session_info |
| 44 | |
| 45 | def write_message(self, message: dict): |
| 46 | self.ioloop.create_task(self.ws.send_json(message)) |
| 47 | |
| 48 | def closed(self) -> bool: |
| 49 | return self.ws.application_state == WebSocketState.DISCONNECTED |
| 50 | |
| 51 | def close(self): |
| 52 | self.ioloop.create_task(self.ws.close()) |
| 53 | |
| 54 | |
| 55 | def _webio_routes(applications, cdn, check_origin_func, reconnect_timeout): |
no outgoing calls
no test coverage detected
searching dependent graphs…