(self)
| 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, |
| 159 | # otherwise, it's self.connection. |
| 160 | if self.reconnectable: |
| 161 | conn = _reconnect_state.active_connections.get(self.session_id) |
| 162 | else: |
| 163 | conn = self.connection |
| 164 | |
| 165 | return conn |
| 166 | |
| 167 | def _send_msg_to_client(self, session: Session = None): |
| 168 | conn = self._get_active_connection() |
no test coverage detected