WebRTCSignalingClient

def __init__(
    self,
    server: str,
    enable_https: bool = False,
    enable_basic_auth: bool = False,
    basic_auth_user: Optional[str] = None,
    basic_auth_password: Optional[str] = None,
    server_token: Optional[str] = None,
) -> None:
    """Initialize the signaling client.

    Args:
        server: WebSocket server URL (e.g., 'ws://localhost:8080/ws').
        server_token: Master token proving this peer may claim the server
            role; required by the signaling server in secure mode.
    """
    self.server = server
    self.peer_type = "server"
    self.enable_https = enable_https
    self.enable_basic_auth = enable_basic_auth
    self.basic_auth_user = basic_auth_user
    self.basic_auth_password = basic_auth_password
    self.server_token = server_token

    self._session: Optional[aiohttp.ClientSession] = None
    self._ws: Optional[ClientWebSocketResponse] = None
    self._stop_event = asyncio.Event()
    self._task: Optional[asyncio.Task] = None

    self.on_ice: Callable[[Dict[str, Any], str], Awaitable[None]] = (
        lambda ice, client_peer_id: logger.warning("unhandled ice event")
    )
    self.on_sdp: Callable[[str, str, str], Awaitable[None]] = (
        lambda sdp_type, sdp, client_peer_id: logger.warning("unhandled sdp event")
    )
    self.on_disconnect: Callable[[], Awaitable[None]] = lambda: logger.warning(
        "unhandled on_disconnect callback"
    )
    self.on_session_start: Callable[..., Awaitable[None]] = (
        lambda client_peer_id, client_type, client_token=None,
        display_id="primary", display_position="right": logger.warning(
            "unhandled on_session_start callback"
        )
    )
    self.on_session_end: Callable[[str, str], Awaitable[None]] = (
        lambda client_peer_id, client_type: logger.warning(
            "unhandled on_session_end callback"
        )
    )
    self.on_error: Callable[[Exception], Awaitable[None]] = lambda v: logger.warning(
        "unhandled on_error callback: %s", v
    )

def start(self) -> None:
    """Start the signaling client connection task."""
    self._stop_event.clear()
    self._task = asyncio.create_task(self.connect_and_listen())

def _hello_message(self) -> str:
    """Registration line for this peer.

    In secure mode the signaling server only lets a peer claim the server
    role when it presents the master token, so carry it in the metadata
    object whenever one is configured.
    """
    if not self.server_token:
        return "HELLO {}".format(self.peer_type)
    metadata = json.dumps({"server_token": self.server_token})
    return "HELLO {} {}".format(self.peer_type, metadata)

async def connect_and_listen(self) -> None:
    """Connect to the signaling server and listen for messages.

    Automatically reconnects on connection failures.
    """
    ssl_ctx: Optional[ssl.SSLContext] = None
    if self.enable_https:
        ssl_ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
        ssl_ctx.check_hostname = False
        ssl_ctx.verify_mode = ssl.CERT_NONE

    headers: Optional[Dict[str, str]] = None
    if self.enable_basic_auth and self.basic_auth_user and self.basic_auth_password:
        # UTF-8 (the server's advertised charset): an ASCII encode of a
        # non-ASCII password would raise here, outside the retry loop.
        auth64 = base64.b64encode(
            f"{self.basic_auth_user}:{self.basic_auth_password}".encode("utf-8")
        ).decode("ascii")
        headers = {"Authorization": f"Basic {auth64}"}

    while not self._stop_event.is_set():
        try:
            logger.info("Connecting to signaling server")
            self._session = aiohttp.ClientSession()
            self._ws = await self._session.ws_connect(
                self.server,
                headers=headers,
                ssl=ssl_ctx,
                heartbeat=30,
            )
            await self._ws.send_str(self._hello_message())
            await self._listen()
        except asyncio.CancelledError:
            pass
        except (
            aiohttp.WSServerHandshakeError,
            aiohttp.ClientConnectionError,
            OSError,
        ) as err:
            logger.warning(f"Connection failed, retrying... {err}")
            await asyncio.sleep(2)
        except aiohttp.ClientError as e:
            logger.warning(f"Client error, attempting to reconnect... {e}")
            await asyncio.sleep(2)
        except Exception as e:
            logger.exception(f"Unexpected error: {e}")
            await asyncio.sleep(2)
        finally:
            await self._cleanup_connection()
            if not self._stop_event.is_set():
                await asyncio.sleep(0.1)

async def _cleanup_connection(self) -> None:
    """Clean up WebSocket connection and session resources."""
    if self._ws is not None and not self._ws.closed:
        await self._ws.close()
    self._ws = None

    if self._session is not None and not self._session.closed:
        await self._session.close()
    self._session = None

async def send_ice(
    self, mlineindex: int, candidate: str, client_peer_id: str
) -> None:
    """Send ICE candidate to peer via signaling server.

    Raises:
        WebRTCSignalingError: If the WebSocket connection is not open.
    """
    if self._ws is None or self._ws.closed:
        raise WebRTCSignalingError("WebSocket connection not available")

    msg = json.dumps({"ice": {"candidate": candidate, "sdpMLineIndex": mlineindex}})
    await self._ws.send_str(f"{client_peer_id} {msg}")

async def send_sdp(self, sdp_type: str, sdp: str, client_peer_id: str) -> None:
    """Send SDP to peer via signaling server.

    Raises:
        WebRTCSignalingError: If the WebSocket connection is not open.
    """
    if self._ws is None or self._ws.closed:
        raise WebRTCSignalingError("WebSocket connection not available")

    logger.info(f"sending sdp type: {sdp_type} to client_peer_id: {client_peer_id}")
    logger.debug("SDP:\n%s" % sdp)

    msg = json.dumps({"sdp": {"type": sdp_type, "sdp": sdp}})
    await self._ws.send_str(f"{client_peer_id} {msg}")

async def stop(self) -> None:
    """Stop the signaling client and clean up resources."""
    logger.info("Stopping signaling client...")
    self._stop_event.set()

    if self._task is not None and not self._task.done():
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass

    await self._cleanup_connection()
    logger.info("Signaling client stopped")

async def _listen(self) -> None:
    """Pump text frames into `_process_message` until the socket closes
    or errors, then fire `on_disconnect`."""
    if self._ws is None:
        raise WebRTCSignalingError("WebSocket connection not available")

    try:
        async for msg in self._ws:
            if msg.type == WSMsgType.TEXT:
                await self._process_message(msg.data)
            elif msg.type == WSMsgType.CLOSED:
                logger.warning("WebSocket connection closed by server")
                break
            elif msg.type == WSMsgType.ERROR:
                logger.error(f"WebSocket error: {self._ws.exception()}")
                break
    except asyncio.CancelledError:
        pass
    except aiohttp.ClientError as e:
        logger.warning(
            f"Signaling server closed the connection: {e}", exc_info=True
        )
    except Exception as e:
        logger.error(f"Error processing signaling message: {e}", exc_info=True)
    finally:
        await self.on_disconnect()

async def _process_message(self, message: str) -> None:
    """Dispatch one signaling line to its callback.

    Lines: `HELLO` (registration ack); `SESSION_START <peer_id>
    <client_type> [<display_id> <display_position> [<client_token>]]`,
    where the 3-token form maps to the primary display and the sixth
    token is the secure-mode collaboration token; `SESSION_END <peer_id>
    <client_type>`; `ERROR ...`; otherwise `<peer_id> <json>` carrying an
    `sdp` or `ice` object. A client's text is relayed here verbatim, so a
    malformed, non-object or unrecognized payload — and a callback that
    raises on a stale SDP/ICE — is logged and dropped rather than treated
    as a transport failure, which would tear down every peer's session.
    """
    if message == "HELLO":
        logger.info("WebSocket connection established with signaling server")

    elif message.startswith("SESSION_START"):
        toks = message.strip().split(" ")
        if len(toks) in (3, 5, 6):
            client_peer_id = toks[1]
            client_type = toks[2]
            display_id = toks[3] if len(toks) >= 5 else "primary"
            display_position = toks[4] if len(toks) >= 5 else "right"
            client_token = toks[5] if len(toks) == 6 else None
            await self.on_session_start(
                client_peer_id, client_type, client_token, display_id, display_position
            )
        else:
            logger.error(f"invalid SESSION_START message: {message}")

    elif message.startswith("SESSION_END"):
        toks = message.strip().split(" ")
        if len(toks) == 3:
            _, client_peer_id, client_type = toks
            await self.on_session_end(client_peer_id, client_type)
        else:
            logger.error(f"invalid SESSION_END message: {message}")

    elif message.startswith("ERROR"):
        await self.on_error(
            WebRTCSignalingError(f"unhandled signaling message: {message}")
        )

    else:
        client_peer_id: Optional[str] = None
        data: Optional[Dict[str, Any]] = None

        try:
            client_peer_id, message = message.split(" ", maxsplit=1)
            data = json.loads(message)
        except ValueError:
            # Covers both a missing peer prefix and JSONDecodeError.
            logger.warning(f"ignoring unparsable signaling message: {message}")
            return

        if not isinstance(data, dict):
            logger.warning(
                f"ignoring non-object JSON signaling message from "
                f"{client_peer_id}: {message}"
            )
            return

        try:
            if isinstance(data.get("sdp"), dict):
                logger.info(f"received SDP from client_peer_id: {client_peer_id}")
                logger.debug(f"SDP:\n{data['sdp']}")
                await self.on_sdp(
                    data["sdp"].get("type", ""),
                    data["sdp"].get("sdp", ""),
                    client_peer_id,
                )
            elif isinstance(data.get("ice"), dict):
                logger.info(f"received ICE from client_peer_id: {client_peer_id}")
                logger.debug(f"ICE:\n{data.get('ice')}")
                await self.on_ice(data["ice"], client_peer_id)
            else:
                logger.warning(
                    f"ignoring unrecognized JSON signaling message from "
                    f"{client_peer_id}: {json.dumps(data)}"
                )
                return
        except Exception as e:
            logger.error(
                f"Error dispatching signaling message from {client_peer_id}: {e}",
                exc_info=True,
            )
            return