| 43 | |
| 44 | |
| 45 | class WebSocketConnection(ws_adaptor.WebSocketConnection): |
| 46 | |
| 47 | def __init__(self, ws: web.WebSocketResponse, http: web.Request, ioloop): |
| 48 | self.ws = ws |
| 49 | self.http = http |
| 50 | self.ioloop = ioloop |
| 51 | |
| 52 | def get_query_argument(self, name) -> typing.Optional[str]: |
| 53 | return self.http.query.getone(name, None) |
| 54 | |
| 55 | def make_session_info(self) -> dict: |
| 56 | session_info = get_session_info_from_headers(self.http.headers) |
| 57 | session_info['user_ip'] = self.http.remote |
| 58 | session_info['request'] = self.http |
| 59 | session_info['backend'] = 'aiohttp' |
| 60 | session_info['protocol'] = 'websocket' |
| 61 | return session_info |
| 62 | |
| 63 | def write_message(self, message: dict): |
| 64 | msg_str = json.dumps(message) |
| 65 | self.ioloop.create_task(self.ws.send_str(msg_str)) |
| 66 | |
| 67 | def closed(self) -> bool: |
| 68 | return self.ws.closed |
| 69 | |
| 70 | def close(self): |
| 71 | self.ioloop.create_task(self.ws.close()) |
| 72 | |
| 73 | |
| 74 | def _webio_handler(applications, cdn, websocket_settings, reconnect_timeout=0, check_origin_func=_is_same_site): |
no outgoing calls
no test coverage detected
searching dependent graphs…