: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, handler) -> bool :return: Tornado RequestHandler class
(applications=None, cdn=True, reconnect_timeout=0, check_origin_func=_is_same_site)
| 92 | |
| 93 | |
| 94 | def _webio_handler(applications=None, cdn=True, reconnect_timeout=0, check_origin_func=_is_same_site) \ |
| 95 | -> tornado.websocket.WebSocketHandler: # noqa: C901 |
| 96 | """ |
| 97 | :param dict applications: dict of `name -> task function` |
| 98 | :param bool/str cdn: Whether to load front-end static resources from CDN |
| 99 | :param callable check_origin_func: check_origin_func(origin, handler) -> bool |
| 100 | :return: Tornado RequestHandler class |
| 101 | """ |
| 102 | check_webio_js() |
| 103 | |
| 104 | if applications is None: |
| 105 | applications = dict(index=lambda: None) # mock PyWebIO app |
| 106 | |
| 107 | ws_adaptor.set_expire_second(reconnect_timeout) |
| 108 | tornado.ioloop.IOLoop.current().spawn_callback(ws_adaptor.session_clean_task) |
| 109 | |
| 110 | class Handler(tornado.websocket.WebSocketHandler): |
| 111 | |
| 112 | def get_app(self): |
| 113 | app_name = self.get_query_argument('app', 'index') |
| 114 | app = applications.get(app_name) or applications['index'] |
| 115 | return app |
| 116 | |
| 117 | def get_cdn(self): |
| 118 | if cdn is True and self.get_query_argument('_pywebio_cdn', '') == 'false': |
| 119 | return False |
| 120 | return cdn |
| 121 | |
| 122 | async def get(self, *args, **kwargs) -> None: |
| 123 | """http GET request""" |
| 124 | if self.request.headers.get("Upgrade", "").lower() != "websocket": |
| 125 | # Backward compatible |
| 126 | # Frontend detect whether the backend is http server |
| 127 | if self.get_query_argument('test', ''): |
| 128 | return self.write('') |
| 129 | |
| 130 | app = self.get_app() |
| 131 | html = render_page(app, protocol='ws', cdn=self.get_cdn()) |
| 132 | return self.write(html) |
| 133 | else: |
| 134 | await super().get() |
| 135 | |
| 136 | def check_origin(self, origin): |
| 137 | return check_origin_func(origin=origin, handler=self) |
| 138 | |
| 139 | def get_compression_options(self): |
| 140 | # Non-None enables compression with default options. |
| 141 | return {} |
| 142 | |
| 143 | _handler: ws_adaptor.WebSocketHandler |
| 144 | |
| 145 | def open(self): |
| 146 | conn = WebSocketConnection(self) |
| 147 | self._handler = ws_adaptor.WebSocketHandler( |
| 148 | connection=conn, application=self.get_app(), reconnectable=bool(reconnect_timeout) |
| 149 | ) |
| 150 | |
| 151 | def on_message(self, message): |
no test coverage detected
searching dependent graphs…