(self, event, namespace, *args)
| 198 | self.stats_task = None |
| 199 | |
| 200 | def _trigger_event(self, event, namespace, *args): |
| 201 | t = time.time() |
| 202 | sid = args[0] |
| 203 | if event == 'connect': |
| 204 | eio_sid = self.sio.manager.eio_sid_from_sid(sid, namespace) |
| 205 | self.sio.manager._timestamps[sid] = t |
| 206 | serialized_socket = self.serialize_socket(sid, namespace, eio_sid) |
| 207 | self.sio.emit('socket_connected', ( |
| 208 | serialized_socket, |
| 209 | datetime.fromtimestamp(t, timezone.utc).isoformat(), |
| 210 | ), namespace=self.admin_namespace) |
| 211 | if not self.sio.eio._get_socket(eio_sid).upgraded: |
| 212 | self.sio.start_background_task( |
| 213 | self._check_for_upgrade, eio_sid, sid, namespace) |
| 214 | elif event == 'disconnect': |
| 215 | del self.sio.manager._timestamps[sid] |
| 216 | reason = args[1] |
| 217 | self.sio.emit('socket_disconnected', ( |
| 218 | namespace, |
| 219 | sid, |
| 220 | reason, |
| 221 | datetime.fromtimestamp(t, timezone.utc).isoformat(), |
| 222 | ), namespace=self.admin_namespace) |
| 223 | else: |
| 224 | self.sio.emit('event_received', ( |
| 225 | namespace, |
| 226 | sid, |
| 227 | (event, *args[1:]), |
| 228 | datetime.fromtimestamp(t, timezone.utc).isoformat(), |
| 229 | ), namespace=self.admin_namespace) |
| 230 | return self.sio.__trigger_event(event, namespace, *args) |
| 231 | |
| 232 | def _check_for_upgrade(self, eio_sid, sid, namespace): # pragma: no cover |
| 233 | for _ in range(5): |
nothing calls this directly
no test coverage detected