:param dict applications: dict of `name -> task function` :param bool/str cdn: Whether to load front-end static resources from CDN :param callable check_origin_func: check_origin_func(origin, host) -> bool
(applications, cdn, check_origin_func, reconnect_timeout)
| 53 | |
| 54 | |
| 55 | def _webio_routes(applications, cdn, check_origin_func, reconnect_timeout): |
| 56 | """ |
| 57 | :param dict applications: dict of `name -> task function` |
| 58 | :param bool/str cdn: Whether to load front-end static resources from CDN |
| 59 | :param callable check_origin_func: check_origin_func(origin, host) -> bool |
| 60 | """ |
| 61 | |
| 62 | ws_adaptor.set_expire_second(reconnect_timeout) |
| 63 | |
| 64 | async def http_endpoint(request: Request): |
| 65 | origin = request.headers.get('origin') |
| 66 | if origin and not check_origin_func(origin=origin, host=request.headers.get('host')): |
| 67 | return HTMLResponse(status_code=403, content="Cross origin websockets not allowed") |
| 68 | |
| 69 | # Backward compatible |
| 70 | if request.query_params.get('test'): |
| 71 | return HTMLResponse(content="") |
| 72 | |
| 73 | app_name = request.query_params.get('app', 'index') |
| 74 | app = applications.get(app_name) or applications['index'] |
| 75 | no_cdn = cdn is True and request.query_params.get('_pywebio_cdn', '') == 'false' |
| 76 | html = render_page(app, protocol='ws', cdn=False if no_cdn else cdn) |
| 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), |
| 109 | WebSocketRoute("/", websocket_endpoint) |
| 110 | ] |
| 111 | |
| 112 |
no outgoing calls
no test coverage detected
searching dependent graphs…