music-assistant-server
35.2 KB•PY
gateway.py
35.2 KB • 844 lines • python
1"""Music Assistant WebRTC Gateway.
2
3This module provides WebRTC-based remote access to Music Assistant instances.
4It connects to a signaling server and handles incoming WebRTC connections,
5bridging them to the local WebSocket API.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import contextlib
12import json
13import logging
14from collections.abc import Awaitable, Callable
15from dataclasses import dataclass, field
16from typing import TYPE_CHECKING, Any
17from urllib.parse import urlparse
18
19import aiohttp
20from aiortc import RTCConfiguration, RTCIceServer, RTCPeerConnection, RTCSessionDescription
21from aiortc.sdp import candidate_from_sdp
22
23from music_assistant.constants import MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
24from music_assistant.helpers.webrtc_certificate import create_peer_connection_with_certificate
25
26if TYPE_CHECKING:
27 from aiortc.rtcdtlstransport import RTCCertificate
28
29LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.remote_access")
30
31# Reduce verbose logging from aiortc/aioice
32logging.getLogger("aioice").setLevel(logging.WARNING)
33logging.getLogger("aiortc").setLevel(logging.WARNING)
34
35
36@dataclass
37class WebRTCSession:
38 """Represents an active WebRTC session with a remote client."""
39
40 session_id: str
41 peer_connection: RTCPeerConnection
42 # Main API channel (ma-api) - bridges to local MA WebSocket API
43 data_channel: Any = None
44 local_ws: Any = None
45 message_queue: asyncio.Queue[str] = field(default_factory=asyncio.Queue)
46 forward_to_local_task: asyncio.Task[None] | None = None
47 forward_from_local_task: asyncio.Task[None] | None = None
48 # Sendspin channel - bridges to internal sendspin server
49 sendspin_channel: Any = None
50 sendspin_ws: Any = None
51 sendspin_queue: asyncio.Queue[str | bytes] = field(default_factory=asyncio.Queue)
52 sendspin_player_id: str | None = None # Extracted from first sendspin auth message
53 sendspin_to_local_task: asyncio.Task[None] | None = None
54 sendspin_from_local_task: asyncio.Task[None] | None = None
55
56
57class WebRTCGateway:
58 """WebRTC Gateway for Music Assistant Remote Access.
59
60 This gateway:
61 1. Connects to a signaling server
62 2. Registers with a unique Remote ID
63 3. Handles incoming WebRTC connections from remote PWA clients
64 4. Bridges WebRTC DataChannel messages to the local WebSocket API
65 """
66
67 # Close code 4000 means this connection was replaced by a new one from the same server
68 # In that case, we should not reconnect as another connection is now active
69 CLOSE_CODE_REPLACED = 4000
70
71 # Default ICE servers (public STUN only - used as fallback)
72 DEFAULT_ICE_SERVERS: list[dict[str, Any]] = [
73 {"urls": "stun:stun.home-assistant.io:3478"},
74 {"urls": "stun:stun.l.google.com:19302"},
75 {"urls": "stun:stun1.l.google.com:19302"},
76 {"urls": "stun:stun.cloudflare.com:3478"},
77 ]
78
79 def __init__(
80 self,
81 http_session: aiohttp.ClientSession,
82 remote_id: str,
83 certificate: RTCCertificate,
84 signaling_url: str = "wss://signaling.music-assistant.io/ws",
85 local_ws_url: str = "ws://localhost:8095/ws",
86 sendspin_url: str = "ws://localhost:8927/sendspin",
87 ice_servers: list[dict[str, Any]] | None = None,
88 ice_servers_callback: Callable[[], Awaitable[list[dict[str, Any]]]] | None = None,
89 set_sendspin_player_callback: Callable[[str, str], None] | None = None,
90 ) -> None:
91 """
92 Initialize the WebRTC Gateway.
93
94 :param http_session: Shared aiohttp ClientSession for HTTP/WebSocket connections.
95 :param remote_id: Remote ID for this server instance.
96 :param certificate: Persistent RTCCertificate for DTLS, enabling client-side pinning.
97 :param signaling_url: WebSocket URL of the signaling server.
98 :param local_ws_url: Local WebSocket URL to bridge to.
99 :param sendspin_url: Internal Sendspin WebSocket URL to bridge to.
100 :param ice_servers: List of ICE server configurations (used at registration time).
101 :param ice_servers_callback: Optional callback to fetch fresh ICE servers for each session.
102 :param set_sendspin_player_callback: Callback to set sendspin player for a session.
103 """
104 self.http_session = http_session
105 self.signaling_url = signaling_url
106 self.local_ws_url = local_ws_url
107 self.sendspin_url = sendspin_url
108 self._remote_id = remote_id
109 self._certificate = certificate
110 self.logger = LOGGER
111 self._ice_servers_callback = ice_servers_callback
112 self._set_sendspin_player_callback = set_sendspin_player_callback
113
114 # Static ICE servers used at registration time (relayed to clients via signaling server)
115 self.ice_servers = ice_servers or self.DEFAULT_ICE_SERVERS
116
117 self.sessions: dict[str, WebRTCSession] = {}
118 self._signaling_ws: aiohttp.ClientWebSocketResponse | None = None
119 self._running = False
120 self._reconnect_delay = 10 # Wait 10 seconds before reconnecting
121 self._max_reconnect_delay = 300 # Max 5 minutes between reconnects
122 self._current_reconnect_delay = 10
123 self._run_task: asyncio.Task[None] | None = None
124 self._is_connected = False
125 self._connecting = False
126
127 @property
128 def is_running(self) -> bool:
129 """Return whether the gateway is running."""
130 return self._running
131
132 @property
133 def is_connected(self) -> bool:
134 """Return whether the gateway is connected to the signaling server."""
135 return self._is_connected
136
137 async def _get_fresh_ice_servers(self) -> list[dict[str, Any]]:
138 """Get fresh ICE servers for a new WebRTC session.
139
140 If an ice_servers_callback was provided, it will be called to get fresh
141 TURN credentials. Otherwise, returns the static ice_servers.
142
143 :return: List of ICE server configurations with fresh credentials.
144 """
145 if self._ice_servers_callback:
146 try:
147 fresh_servers = await self._ice_servers_callback()
148 if fresh_servers:
149 return fresh_servers
150 except Exception:
151 self.logger.exception("Failed to fetch fresh ICE servers, using cached servers")
152 return self.ice_servers
153
154 async def start(self) -> None:
155 """Start the WebRTC Gateway."""
156 if self._running:
157 self.logger.warning("WebRTC Gateway already running, skipping start")
158 return
159 self.logger.info("Starting WebRTC Gateway")
160 self.logger.debug("Signaling URL: %s", self.signaling_url)
161 self.logger.debug("Local WS URL: %s", self.local_ws_url)
162 self._running = True
163 self._run_task = asyncio.create_task(self._run())
164 self.logger.debug("WebRTC Gateway start task created")
165
166 async def stop(self) -> None:
167 """Stop the WebRTC Gateway."""
168 self.logger.info("Stopping WebRTC Gateway")
169 self._running = False
170
171 # Close all sessions
172 for session_id in list(self.sessions.keys()):
173 await self._close_session(session_id)
174
175 # Close signaling connection gracefully
176 if self._signaling_ws and not self._signaling_ws.closed:
177 try:
178 await self._signaling_ws.close()
179 except Exception:
180 self.logger.debug("Error closing signaling WebSocket", exc_info=True)
181
182 # Cancel run task and wait for it to finish
183 if self._run_task and not self._run_task.done():
184 self._run_task.cancel()
185 with contextlib.suppress(asyncio.CancelledError):
186 await self._run_task
187
188 # Wait briefly for any in-progress connection to notice _running=False
189 if self._connecting:
190 await asyncio.sleep(0.1)
191
192 self._signaling_ws = None
193 self._connecting = False
194
195 async def _run(self) -> None:
196 """Run the main loop with reconnection logic."""
197 self.logger.debug("WebRTC Gateway _run() loop starting")
198 while self._running:
199 should_reconnect = True
200 try:
201 should_reconnect = await self._connect_to_signaling()
202 # Connection closed gracefully or with error
203 self._is_connected = False
204 if self._running and should_reconnect:
205 self.logger.warning(
206 "Signaling server connection lost. Reconnecting in %ss...",
207 self._current_reconnect_delay,
208 )
209 except Exception:
210 self._is_connected = False
211 self.logger.exception("Signaling connection error")
212 if self._running:
213 self.logger.info(
214 "Reconnecting to signaling server in %ss",
215 self._current_reconnect_delay,
216 )
217
218 if self._running and should_reconnect:
219 await asyncio.sleep(self._current_reconnect_delay)
220 # Exponential backoff with max limit
221 self._current_reconnect_delay = min(
222 self._current_reconnect_delay * 2, self._max_reconnect_delay
223 )
224 elif not should_reconnect:
225 # Connection was replaced by another instance, stop the run loop
226 self.logger.info("Connection replaced, stopping reconnection attempts")
227 self._running = False
228 break
229
230 async def _connect_to_signaling(self) -> bool:
231 """Connect to the signaling server.
232
233 :return: True if reconnection should be attempted, False if connection was replaced.
234 """
235 if self._connecting:
236 self.logger.warning("Already connecting to signaling server, skipping")
237 return False # Don't trigger another reconnect cycle
238 self._connecting = True
239 close_code: int | None = None
240 self.logger.info("Connecting to signaling server: %s", self.signaling_url)
241 try:
242 self._signaling_ws = await self.http_session.ws_connect(
243 self.signaling_url,
244 heartbeat=35.0, # Send ping every 35s (slightly above server's 30s interval)
245 )
246 # Check if we were stopped while connecting
247 if not self._running:
248 self.logger.debug("Gateway stopped during connection, closing WebSocket")
249 await self._signaling_ws.close()
250 self._signaling_ws = None
251 self._connecting = False
252 return False
253 self.logger.debug("WebSocket connection established, id=%s", id(self._signaling_ws))
254 self.logger.debug("Sending registration")
255 await self._register()
256 self._current_reconnect_delay = self._reconnect_delay
257 self.logger.debug("Registration sent, waiting for confirmation...")
258
259 # Run message loop and get close code
260 close_code = await self._signaling_message_loop(self._signaling_ws)
261
262 # Get close code from WebSocket if not already set from CLOSE message
263 if close_code is None:
264 close_code = self._signaling_ws.close_code
265 ws_exception = self._signaling_ws.exception()
266 self.logger.debug(
267 "Message loop exited - WebSocket closed: %s, close_code: %s, exception: %s",
268 self._signaling_ws.closed,
269 close_code,
270 ws_exception,
271 )
272 except TimeoutError:
273 self.logger.error("Timeout connecting to signaling server")
274 except aiohttp.ClientError as err:
275 self.logger.error("Failed to connect to signaling server: %s", err)
276 except Exception:
277 self.logger.exception("Unexpected error in signaling connection")
278 finally:
279 self._is_connected = False
280 self._connecting = False
281 self._signaling_ws = None
282
283 # Check if this connection was replaced by another one
284 if close_code == self.CLOSE_CODE_REPLACED:
285 self.logger.info("Connection was replaced by another instance - not reconnecting")
286 return False
287
288 return True
289
290 async def _signaling_message_loop(self, ws: aiohttp.ClientWebSocketResponse) -> int | None:
291 """Process messages from the signaling WebSocket.
292
293 :param ws: The WebSocket connection to process messages from.
294 :return: Close code if connection was closed with a code, None otherwise.
295 """
296 close_code: int | None = None
297 self.logger.debug("Entering message loop")
298 async for msg in ws:
299 if msg.type == aiohttp.WSMsgType.TEXT:
300 try:
301 await self._handle_signaling_message(json.loads(msg.data))
302 except Exception:
303 self.logger.exception("Error handling signaling message")
304 elif msg.type == aiohttp.WSMsgType.PING:
305 self.logger.log(VERBOSE_LOG_LEVEL, "Received WebSocket PING")
306 elif msg.type == aiohttp.WSMsgType.PONG:
307 self.logger.log(VERBOSE_LOG_LEVEL, "Received WebSocket PONG")
308 elif msg.type == aiohttp.WSMsgType.CLOSE:
309 close_code = msg.data
310 self.logger.warning(
311 "Signaling server sent close frame: code=%s, reason=%s",
312 msg.data,
313 msg.extra,
314 )
315 break
316 elif msg.type == aiohttp.WSMsgType.CLOSED:
317 self.logger.warning("Signaling server closed connection")
318 break
319 elif msg.type == aiohttp.WSMsgType.ERROR:
320 self.logger.error("WebSocket error: %s", ws.exception())
321 break
322 else:
323 self.logger.warning("Unexpected WebSocket message type: %s", msg.type)
324 return close_code
325
326 async def _register(self) -> None:
327 """Register with the signaling server."""
328 if self._signaling_ws:
329 await self._signaling_ws.send_json(
330 {
331 "type": "register-server",
332 "remoteId": self._remote_id,
333 "iceServers": self.ice_servers,
334 }
335 )
336
337 async def _handle_signaling_message(self, message: dict[str, Any]) -> None:
338 """Handle incoming signaling messages.
339
340 :param message: The signaling message.
341 """
342 msg_type = message.get("type")
343
344 if msg_type in ("ping", "pong"):
345 # Ignore JSON-level ping/pong messages - we use WebSocket protocol-level heartbeat
346 # The signaling server still sends these for backward compatibility with older clients
347 pass
348 elif msg_type == "registered":
349 self._is_connected = True
350 self.logger.info("Registered with signaling server")
351 elif msg_type == "error":
352 error_msg = message.get("error") or message.get("message", "Unknown error")
353 self.logger.error("Signaling server error: %s", error_msg)
354 elif msg_type == "client-connected":
355 session_id = message.get("sessionId")
356 if session_id:
357 await self._create_session(session_id)
358 # Send session-ready with fresh ICE servers for the client
359 fresh_ice_servers = await self._get_fresh_ice_servers()
360 if self._signaling_ws:
361 await self._signaling_ws.send_json(
362 {
363 "type": "session-ready",
364 "sessionId": session_id,
365 "iceServers": fresh_ice_servers,
366 }
367 )
368 elif msg_type == "client-disconnected":
369 session_id = message.get("sessionId")
370 if session_id:
371 await self._close_session(session_id)
372 elif msg_type == "offer":
373 session_id = message.get("sessionId")
374 offer_data = message.get("data")
375 if session_id and offer_data:
376 await self._handle_offer(session_id, offer_data)
377 elif msg_type == "ice-candidate":
378 session_id = message.get("sessionId")
379 candidate_data = message.get("data")
380 if session_id and candidate_data:
381 await self._handle_ice_candidate(session_id, candidate_data)
382
383 async def _create_session(self, session_id: str) -> None:
384 """Create a new WebRTC session.
385
386 :param session_id: The session ID.
387 """
388 session_ice_servers = await self._get_fresh_ice_servers()
389 config = RTCConfiguration(
390 iceServers=[RTCIceServer(**server) for server in session_ice_servers]
391 )
392 pc = create_peer_connection_with_certificate(self._certificate, configuration=config)
393 session = WebRTCSession(session_id=session_id, peer_connection=pc)
394 self.sessions[session_id] = session
395
396 @pc.on("datachannel")
397 def on_datachannel(channel: Any) -> None:
398 if channel.label == "sendspin":
399 session.sendspin_channel = channel
400 asyncio.create_task(self._setup_sendspin_channel(session))
401 else:
402 session.data_channel = channel
403 asyncio.create_task(self._setup_data_channel(session))
404
405 @pc.on("icecandidate")
406 async def on_icecandidate(candidate: Any) -> None:
407 if candidate and self._signaling_ws:
408 await self._signaling_ws.send_json(
409 {
410 "type": "ice-candidate",
411 "sessionId": session_id,
412 "data": {
413 "candidate": candidate.candidate,
414 "sdpMid": candidate.sdpMid,
415 "sdpMLineIndex": candidate.sdpMLineIndex,
416 },
417 }
418 )
419
420 @pc.on("connectionstatechange")
421 async def on_connectionstatechange() -> None:
422 if pc.connectionState == "failed":
423 await self._close_session(session_id)
424
425 async def _handle_offer(self, session_id: str, offer: dict[str, Any]) -> None:
426 """Handle incoming WebRTC offer.
427
428 :param session_id: The session ID.
429 :param offer: The offer data.
430 """
431 session = self.sessions.get(session_id)
432 if not session:
433 return
434 pc = session.peer_connection
435
436 if pc.connectionState in ("closed", "failed"):
437 return
438
439 sdp = offer.get("sdp")
440 sdp_type = offer.get("type")
441 if not sdp or not sdp_type:
442 self.logger.error("Invalid offer data: missing sdp or type")
443 return
444
445 try:
446 await pc.setRemoteDescription(
447 RTCSessionDescription(
448 sdp=str(sdp),
449 type=str(sdp_type),
450 )
451 )
452
453 if session_id not in self.sessions or pc.connectionState in ("closed", "failed"):
454 return
455
456 answer = await pc.createAnswer()
457
458 if session_id not in self.sessions or pc.connectionState in ("closed", "failed"):
459 return
460
461 await pc.setLocalDescription(answer)
462
463 # Wait for ICE gathering to complete before sending the answer
464 # aiortc doesn't support trickle ICE, candidates are embedded in SDP after gathering
465 gather_timeout = 30
466 gather_start = asyncio.get_event_loop().time()
467 while pc.iceGatheringState != "complete":
468 if session_id not in self.sessions or pc.connectionState in ("closed", "failed"):
469 return
470 if asyncio.get_event_loop().time() - gather_start > gather_timeout:
471 self.logger.warning("Session %s ICE gathering timeout", session_id)
472 break
473 await asyncio.sleep(0.1)
474
475 if session_id not in self.sessions or pc.connectionState in ("closed", "failed"):
476 return
477
478 if self._signaling_ws:
479 await self._signaling_ws.send_json(
480 {
481 "type": "answer",
482 "sessionId": session_id,
483 "data": {
484 "sdp": pc.localDescription.sdp,
485 "type": pc.localDescription.type,
486 },
487 }
488 )
489 except Exception:
490 self.logger.exception("Error handling offer for session %s", session_id)
491 # Clean up the session on error
492 await self._close_session(session_id)
493
494 async def _handle_ice_candidate(self, session_id: str, candidate: dict[str, Any]) -> None:
495 """Handle incoming ICE candidate.
496
497 :param session_id: The session ID.
498 :param candidate: The ICE candidate data.
499 """
500 session = self.sessions.get(session_id)
501 if not session or not candidate:
502 return
503
504 pc = session.peer_connection
505 if pc.connectionState in ("closed", "failed"):
506 return
507
508 candidate_str = candidate.get("candidate")
509 sdp_mid = candidate.get("sdpMid")
510 sdp_mline_index = candidate.get("sdpMLineIndex")
511
512 if not candidate_str:
513 return
514
515 try:
516 # Parse ICE candidate - browser sends "candidate:..." format
517 if candidate_str.startswith("candidate:"):
518 sdp_candidate_str = candidate_str[len("candidate:") :]
519 else:
520 sdp_candidate_str = candidate_str
521
522 ice_candidate = candidate_from_sdp(sdp_candidate_str)
523 ice_candidate.sdpMid = str(sdp_mid) if sdp_mid else None
524 ice_candidate.sdpMLineIndex = (
525 int(sdp_mline_index) if sdp_mline_index is not None else None
526 )
527
528 if session_id not in self.sessions or pc.connectionState in ("closed", "failed"):
529 return
530
531 await session.peer_connection.addIceCandidate(ice_candidate)
532 except Exception:
533 self.logger.exception("Failed to add ICE candidate for session %s", session_id)
534
535 async def _setup_data_channel(self, session: WebRTCSession) -> None:
536 """Set up data channel and bridge to local WebSocket.
537
538 :param session: The WebRTC session.
539 """
540 channel = session.data_channel
541 if not channel:
542 return
543 try:
544 # Include session_id in URL so server can track WebRTC sessions
545 ws_url = f"{self.local_ws_url}?webrtc_session_id={session.session_id}"
546 session.local_ws = await self.http_session.ws_connect(ws_url)
547 loop = asyncio.get_event_loop()
548
549 # Store task references for proper cleanup
550 session.forward_to_local_task = asyncio.create_task(self._forward_to_local(session))
551 session.forward_from_local_task = asyncio.create_task(self._forward_from_local(session))
552
553 @channel.on("message") # type: ignore[untyped-decorator]
554 def on_message(message: str) -> None:
555 # Called from aiortc thread, use call_soon_threadsafe
556 # Only queue message if session is still active
557 if session.forward_to_local_task and not session.forward_to_local_task.done():
558 loop.call_soon_threadsafe(session.message_queue.put_nowait, message)
559
560 @channel.on("close") # type: ignore[untyped-decorator]
561 def on_close() -> None:
562 # Called from aiortc thread, use call_soon_threadsafe to schedule task
563 asyncio.run_coroutine_threadsafe(self._close_session(session.session_id), loop)
564
565 except Exception:
566 self.logger.exception("Failed to connect to local WebSocket")
567
568 async def _forward_to_local(self, session: WebRTCSession) -> None:
569 """Forward messages from WebRTC DataChannel to local WebSocket.
570
571 :param session: The WebRTC session.
572 """
573 try:
574 while session.local_ws and not session.local_ws.closed:
575 message = await session.message_queue.get()
576
577 # Check if this is an HTTP proxy request
578 try:
579 msg_data = json.loads(message)
580 if isinstance(msg_data, dict) and msg_data.get("type") == "http-proxy-request":
581 # Handle HTTP proxy request
582 await self._handle_http_proxy_request(session, msg_data)
583 continue
584 except (json.JSONDecodeError, ValueError):
585 pass
586
587 # Regular WebSocket message
588 if session.local_ws and not session.local_ws.closed:
589 await session.local_ws.send_str(message)
590 except asyncio.CancelledError:
591 # Task was cancelled during cleanup, this is expected
592 self.logger.debug("Forward to local task cancelled for session %s", session.session_id)
593 raise
594 except Exception:
595 self.logger.exception("Error forwarding to local WebSocket")
596
597 async def _forward_from_local(self, session: WebRTCSession) -> None:
598 """Forward messages from local WebSocket to WebRTC DataChannel.
599
600 :param session: The WebRTC session.
601 """
602 try:
603 async for msg in session.local_ws:
604 if msg.type == aiohttp.WSMsgType.TEXT:
605 if session.data_channel and session.data_channel.readyState == "open":
606 session.data_channel.send(msg.data)
607 elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED):
608 break
609 except asyncio.CancelledError:
610 # Task was cancelled during cleanup, this is expected
611 self.logger.debug(
612 "Forward from local task cancelled for session %s", session.session_id
613 )
614 raise
615 except Exception:
616 self.logger.exception("Error forwarding from local WebSocket")
617
618 async def _handle_http_proxy_request(
619 self, session: WebRTCSession, request_data: dict[str, Any]
620 ) -> None:
621 """Handle HTTP proxy request from remote client.
622
623 :param session: The WebRTC session.
624 :param request_data: The HTTP proxy request data.
625 """
626 request_id = request_data.get("id")
627 method = request_data.get("method", "GET")
628 path = request_data.get("path", "/")
629 headers = request_data.get("headers", {})
630
631 # Build local HTTP URL from the WebSocket URL.
632 # Handle both ws:// and wss:// schemes.
633 parsed = urlparse(self.local_ws_url)
634 http_scheme = "https" if parsed.scheme == "wss" else "http"
635 local_http_url = f"{http_scheme}://{parsed.netloc}{path}"
636
637 self.logger.debug("HTTP proxy request: %s %s", method, local_http_url)
638
639 try:
640 # Use shared HTTP session for this request
641 async with self.http_session.request(
642 method, local_http_url, headers=headers
643 ) as response:
644 # Read response body
645 body = await response.read()
646
647 # Prepare response data
648 response_data = {
649 "type": "http-proxy-response",
650 "id": request_id,
651 "status": response.status,
652 "headers": dict(response.headers),
653 "body": body.hex(), # Send as hex string to avoid encoding issues
654 }
655
656 # Send response back through data channel
657 if session.data_channel and session.data_channel.readyState == "open":
658 session.data_channel.send(json.dumps(response_data))
659
660 except Exception as err:
661 self.logger.exception("Error handling HTTP proxy request")
662 # Send error response
663 error_response = {
664 "type": "http-proxy-response",
665 "id": request_id,
666 "status": 500,
667 "headers": {"Content-Type": "text/plain"},
668 "body": str(err).encode().hex(),
669 }
670 if session.data_channel and session.data_channel.readyState == "open":
671 session.data_channel.send(json.dumps(error_response))
672
673 async def _close_session(self, session_id: str) -> None:
674 """Close a WebRTC session.
675
676 :param session_id: The session ID.
677 """
678 session = self.sessions.pop(session_id, None)
679 if not session:
680 return
681
682 # Cancel forwarding tasks first to prevent race conditions
683 if session.forward_to_local_task and not session.forward_to_local_task.done():
684 session.forward_to_local_task.cancel()
685 with contextlib.suppress(asyncio.CancelledError):
686 await session.forward_to_local_task
687
688 if session.forward_from_local_task and not session.forward_from_local_task.done():
689 session.forward_from_local_task.cancel()
690 with contextlib.suppress(asyncio.CancelledError):
691 await session.forward_from_local_task
692
693 # Cancel sendspin forwarding tasks
694 if session.sendspin_to_local_task and not session.sendspin_to_local_task.done():
695 session.sendspin_to_local_task.cancel()
696 with contextlib.suppress(asyncio.CancelledError):
697 await session.sendspin_to_local_task
698
699 if session.sendspin_from_local_task and not session.sendspin_from_local_task.done():
700 session.sendspin_from_local_task.cancel()
701 with contextlib.suppress(asyncio.CancelledError):
702 await session.sendspin_from_local_task
703
704 # Close connections
705 if session.local_ws and not session.local_ws.closed:
706 await session.local_ws.close()
707 if session.sendspin_ws and not session.sendspin_ws.closed:
708 await session.sendspin_ws.close()
709 if session.data_channel:
710 session.data_channel.close()
711 if session.sendspin_channel:
712 session.sendspin_channel.close()
713 await session.peer_connection.close()
714
715 async def _setup_sendspin_channel(self, session: WebRTCSession) -> None:
716 """Set up sendspin data channel and bridge to internal sendspin server.
717
718 :param session: The WebRTC session.
719 """
720 channel = session.sendspin_channel
721 if not channel:
722 return
723
724 try:
725 loop = asyncio.get_event_loop()
726
727 @channel.on("message") # type: ignore[untyped-decorator]
728 def on_message(message: str | bytes) -> None:
729 # Queue if task not yet created (None) or still running.
730 # Only drop when task exists and is done (shutdown).
731 if (
732 session.sendspin_to_local_task is None
733 or not session.sendspin_to_local_task.done()
734 ):
735 loop.call_soon_threadsafe(session.sendspin_queue.put_nowait, message)
736
737 @channel.on("close") # type: ignore[untyped-decorator]
738 def on_close() -> None:
739 if session.sendspin_ws and not session.sendspin_ws.closed:
740 asyncio.run_coroutine_threadsafe(session.sendspin_ws.close(), loop)
741
742 session.sendspin_ws = await self.http_session.ws_connect(self.sendspin_url)
743 self.logger.debug("Sendspin channel connected for session %s", session.session_id)
744
745 # Start forwarding tasks - queued messages will be processed
746 session.sendspin_to_local_task = asyncio.create_task(
747 self._forward_sendspin_to_local(session)
748 )
749 session.sendspin_from_local_task = asyncio.create_task(
750 self._forward_sendspin_from_local(session)
751 )
752
753 except Exception:
754 self.logger.exception(
755 "Failed to connect sendspin channel to internal server for session %s",
756 session.session_id,
757 )
758 # Clean up partial state on failure
759 if session.sendspin_to_local_task:
760 session.sendspin_to_local_task.cancel()
761 if session.sendspin_from_local_task:
762 session.sendspin_from_local_task.cancel()
763 if session.sendspin_ws and not session.sendspin_ws.closed:
764 await session.sendspin_ws.close()
765
766 async def _forward_sendspin_to_local(self, session: WebRTCSession) -> None:
767 """Forward messages from sendspin DataChannel to internal sendspin server.
768
769 :param session: The WebRTC session.
770 """
771 first_message = True
772 try:
773 while session.sendspin_ws and not session.sendspin_ws.closed:
774 message = await session.sendspin_queue.get()
775
776 # Check only the first message for client_id extraction
777 if first_message:
778 first_message = False
779 if isinstance(message, str):
780 self._try_extract_sendspin_client_id(session, message)
781
782 if session.sendspin_ws and not session.sendspin_ws.closed:
783 if isinstance(message, bytes):
784 await session.sendspin_ws.send_bytes(message)
785 else:
786 await session.sendspin_ws.send_str(message)
787 except asyncio.CancelledError:
788 self.logger.debug(
789 "Sendspin forward to local task cancelled for session %s",
790 session.session_id,
791 )
792 raise
793 except Exception:
794 self.logger.exception("Error forwarding sendspin to local")
795
796 def _try_extract_sendspin_client_id(self, session: WebRTCSession, message: str) -> None:
797 """Try to extract client_id from sendspin auth message and set on websocket client.
798
799 :param session: The WebRTC session.
800 :param message: The first sendspin message (expected to be auth).
801 """
802 try:
803 data = json.loads(message)
804 if data.get("type") != "auth":
805 return # Not an auth message
806
807 # This is an auth message - extract client_id if present
808 if client_id := data.get("client_id"):
809 session.sendspin_player_id = client_id
810 self.logger.debug(
811 "Extracted sendspin player %s for session %s",
812 client_id,
813 session.session_id,
814 )
815 # Use callback to set sendspin player on the websocket client
816 if self._set_sendspin_player_callback:
817 self._set_sendspin_player_callback(session.session_id, client_id)
818 except (json.JSONDecodeError, TypeError):
819 pass # Not valid JSON, ignore
820
821 async def _forward_sendspin_from_local(self, session: WebRTCSession) -> None:
822 """Forward messages from internal sendspin server to sendspin DataChannel.
823
824 :param session: The WebRTC session.
825 """
826 if not session.sendspin_ws or session.sendspin_ws.closed:
827 return
828
829 try:
830 async for msg in session.sendspin_ws:
831 if msg.type in {aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY}:
832 if session.sendspin_channel and session.sendspin_channel.readyState == "open":
833 session.sendspin_channel.send(msg.data)
834 elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED):
835 break
836 except asyncio.CancelledError:
837 self.logger.debug(
838 "Sendspin forward from local task cancelled for session %s",
839 session.session_id,
840 )
841 raise
842 except Exception:
843 self.logger.exception("Error forwarding sendspin from local")
844