(self, data: dict[str, Any], exclude_user_id: list[str] | None = None)
| 58 | return success |
| 59 | |
| 60 | async def broadcast(self, data: dict[str, Any], exclude_user_id: list[str] | None = None) -> int: |
| 61 | if exclude_user_id is None: |
| 62 | exclude_user_id = [] |
| 63 | success = 0 |
| 64 | user_ids = [] |
| 65 | for user_id in self._connections.keys(): |
| 66 | if user_id in exclude_user_id: |
| 67 | continue |
| 68 | success += await self.send_to_user(user_id, data) |
| 69 | user_ids.append(user_id) |
| 70 | self.log.info(f"🍐 WebSocket sent messages to {success} connections for users {user_ids}") |
| 71 | return success |
| 72 | |
| 73 | async def _safe_send(self, ws: tuple[WebSocket, asyncio.Lock], data: dict[str, Any], user_id: str) -> bool: |
| 74 | try: |
nothing calls this directly
no test coverage detected