(self, event, data, namespace, room=None, skip_sid=None,
callback=None, **kwargs)
| 266 | return self.sio.manager.__basic_leave_room(sid, namespace, room) |
| 267 | |
| 268 | def _emit(self, event, data, namespace, room=None, skip_sid=None, |
| 269 | callback=None, **kwargs): |
| 270 | ret = self.sio.manager.__emit(event, data, namespace, room=room, |
| 271 | skip_sid=skip_sid, callback=callback, |
| 272 | **kwargs) |
| 273 | if namespace != self.admin_namespace: |
| 274 | event_data = [event] + list(data) if isinstance(data, tuple) \ |
| 275 | else [event, data] |
| 276 | if not isinstance(skip_sid, list): # pragma: no branch |
| 277 | skip_sid = [skip_sid] |
| 278 | for sid, _ in self.sio.manager.get_participants(namespace, room): |
| 279 | if sid not in skip_sid: |
| 280 | self.sio.emit('event_sent', ( |
| 281 | namespace, |
| 282 | sid, |
| 283 | event_data, |
| 284 | datetime.now(timezone.utc).isoformat(), |
| 285 | ), namespace=self.admin_namespace) |
| 286 | return ret |
| 287 | |
| 288 | def _handle_eio_connect(self, eio_sid, environ): |
| 289 | if self.stop_stats_event is None: |
nothing calls this directly
no test coverage detected