called when every http request
(self, context: HttpContext)
| 250 | |
| 251 | @contextmanager |
| 252 | def handle_request_context(self, context: HttpContext): |
| 253 | """called when every http request""" |
| 254 | cls = type(self) |
| 255 | |
| 256 | if _event_loop: |
| 257 | asyncio.set_event_loop(_event_loop) |
| 258 | |
| 259 | request_headers = context.request_headers() |
| 260 | |
| 261 | # CORS process start ############################ |
| 262 | if context.request_method() == 'OPTIONS': # preflight request for CORS |
| 263 | self._process_cors(context) |
| 264 | context.set_status(204) |
| 265 | return context.get_response() |
| 266 | |
| 267 | if request_headers.get('Origin'): # set headers for CORS request |
| 268 | self._process_cors(context) |
| 269 | # CORS process end ############################ |
| 270 | |
| 271 | if context.request_url_parameter('test'): # 测试接口，当会话使用基于http的backend时，返回 ok |
| 272 | context.set_content('ok') |
| 273 | return context.get_response() |
| 274 | |
| 275 | # 对首页HTML的请求 |
| 276 | if 'webio-session-id' not in request_headers: |
| 277 | app = self.app_loader(context) |
| 278 | html = render_page(app, protocol='http', cdn=self.get_cdn(context)) |
| 279 | context.set_content(html) |
| 280 | return context.get_response() |
| 281 | |
| 282 | ack = int(context.request_url_parameter('ack', 0)) |
| 283 | webio_session_id = request_headers['webio-session-id'] |
| 284 | new_request = False |
| 285 | if webio_session_id.startswith('NEW-'): |
| 286 | new_request = True |
| 287 | webio_session_id = webio_session_id[4:] |
| 288 | |
| 289 | if new_request and webio_session_id not in cls._webio_sessions: # 初始请求，创建新 Session |
| 290 | if context.request_method() == 'POST': # 不能在POST请求中创建Session，防止CSRF攻击 |
| 291 | context.set_status(403) |
| 292 | return context.get_response() |
| 293 | |
| 294 | session_info = get_session_info_from_headers(context.request_headers()) |
| 295 | session_info['user_ip'] = context.get_client_ip() |
| 296 | session_info['request'] = context.request_obj() |
| 297 | session_info['backend'] = context.backend_name |
| 298 | session_info['protocol'] = 'http' |
| 299 | |
| 300 | application = self.app_loader(context) |
| 301 | |
| 302 | if iscoroutinefunction(application) or isgeneratorfunction(application): |
| 303 | session_cls = CoroutineBasedSession |
| 304 | else: |
| 305 | session_cls = ThreadBasedSession |
| 306 | webio_session = session_cls(application, session_info=session_info) |
| 307 | cls._webio_sessions[webio_session_id] = webio_session |
| 308 | cls._webio_transports[webio_session_id] = ReliableTransport(webio_session) |
| 309 | yield cls.WAIT_MS_ON_POST / 1000.0 # <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- |
no test coverage detected