(websocket: WebSocket)
| 77 | return HTMLResponse(content=html) |
| 78 | |
| 79 | async def websocket_endpoint(websocket: WebSocket): |
| 80 | ioloop = asyncio.get_event_loop() |
| 81 | asyncio.get_event_loop().create_task(ws_adaptor.session_clean_task()) |
| 82 | |
| 83 | await websocket.accept() |
| 84 | |
| 85 | app_name = websocket.query_params.get('app', 'index') |
| 86 | application = applications.get(app_name) or applications['index'] |
| 87 | |
| 88 | conn = WebSocketConnection(websocket, ioloop) |
| 89 | handler = ws_adaptor.WebSocketHandler( |
| 90 | connection=conn, application=application, reconnectable=bool(reconnect_timeout), ioloop=ioloop |
| 91 | ) |
| 92 | |
| 93 | while True: |
| 94 | try: |
| 95 | msg = await websocket.receive() |
| 96 | if msg["type"] == "websocket.disconnect": |
| 97 | raise WebSocketDisconnect(msg["code"]) |
| 98 | text, binary = msg.get('text'), msg.get('bytes') |
| 99 | if text: |
| 100 | handler.send_client_data(text) |
| 101 | if binary: |
| 102 | handler.send_client_data(binary) |
| 103 | except WebSocketDisconnect: |
| 104 | handler.notify_connection_lost() |
| 105 | break |
| 106 | |
| 107 | return [ |
| 108 | Route("/", http_endpoint), |
nothing calls this directly
no test coverage detected
searching dependent graphs…