(self, application)
| 139 | logger.debug('session id: %s' % self.session_id) |
| 140 | |
| 141 | def _init_session(self, application): |
| 142 | session_info = self.connection.make_session_info() |
| 143 | self.session_id = random_str(24) |
| 144 | |
| 145 | if iscoroutinefunction(application) or isgeneratorfunction(application): |
| 146 | self.session = CoroutineBasedSession( |
| 147 | application, session_info=session_info, |
| 148 | on_task_command=self._send_msg_to_client, |
| 149 | on_session_close=self._close_from_session) |
| 150 | else: |
| 151 | self.session = ThreadBasedSession( |
| 152 | application, session_info=session_info, |
| 153 | on_task_command=self._send_msg_to_client, |
| 154 | on_session_close=self._close_from_session, |
| 155 | loop=self.ioloop) |
| 156 | |
| 157 | def _get_active_connection(self) -> Optional[WebSocketConnection]: |
| 158 | # when reconnect enabled, the active connection for this session is in _reconnect_state.active_connections, |
no test coverage detected