fix: re-add connection logic

Author: maxkahan

examples/stt_deepgram_transcription/main.py

@@ -85,15 +85,17 @@ async def on_audio(pcm: PcmData, user):
             @stt.on("transcript")
             async def on_transcript(text: str, user: any, metadata: dict):
                 timestamp = time.strftime("%H:%M:%S")
-                user_info = user if user else "unknown"
+                user_info = user.name if user and hasattr(user, "name") else "unknown"
                 print(f"[{timestamp}] {user_info}: {text}")
                 if metadata.get("confidence"):
                     print(f"    └─ confidence: {metadata['confidence']:.2%}")
 
             @stt.on("partial_transcript")
             async def on_partial_transcript(text: str, user: any, metadata: dict):
                 if text.strip():  # Only show non-empty partial transcripts
-                    user_info = user if user else "unknown"
+                    user_info = (
+                        user.name if user and hasattr(user, "name") else "unknown"
+                    )
                     print(
                         f"    {user_info} (partial): {text}", end="\r"
                     )  # Overwrite line

getstream/video/rtc/connection_manager.py

@@ -9,8 +9,8 @@
 from twirp.context import Context
 
 from getstream.utils import StreamAsyncIOEventEmitter
-from getstream.video.rtc.coordinator import StreamAPIWS
 from getstream.video.rtc.pb.stream.video.sfu.models import models_pb2
+from getstream.video.rtc.pb.stream.video.sfu.signal_rpc import signal_pb2
 from getstream.video.rtc.twirp_client_wrapper import SignalClient
 
 from getstream.video.call import Call
@@ -39,7 +39,7 @@ async def _log_event(event_type: str, data: Any):
 
 class ConnectionManager(StreamAsyncIOEventEmitter):
     """Main connection manager facade for video streaming."""
-    
+
     def __init__(
         self,
         call: Call,
@@ -59,19 +59,22 @@ def __init__(
         self.session_id: str = str(uuid.uuid4())
         self.join_response: Optional[JoinCallResponse] = None
         self.local_sfu: bool = False  # Local SFU flag for development
-        
+
         # Private attributes
         self._connection_state: ConnectionState = ConnectionState.IDLE
         self._stop_event: asyncio.Event = asyncio.Event()
         self._connection_options: ConnectionOptions = ConnectionOptions()
         self._ws_client = None
-        
+        self._coordinator_ws_client = None
+
         # Initialize private managers
         self._participants_state: ParticipantsState = ParticipantsState()
         self._recording_manager: RecordingManager = RecordingManager()
         self._network_monitor: NetworkMonitor = NetworkMonitor(self)
         self._reconnector: ReconnectionManager = ReconnectionManager(self)
-        self._subscription_manager: SubscriptionManager = SubscriptionManager(self, subscription_config)
+        self._subscription_manager: SubscriptionManager = SubscriptionManager(
+            self, subscription_config
+        )
         self._peer_manager: PeerConnectionManager = PeerConnectionManager(self)
 
         self.recording_manager = self._recording_manager  # type: ignore
@@ -93,11 +96,8 @@ def connection_state(self, state: ConnectionState):
             old_state = self._connection_state
             self._connection_state = state
             # Schedule the emit as a background task since property setters cannot be async
-            self.emit('connection.state_changed', {
-                'old': old_state,
-                'new': state
-            })
-    
+            self.emit("connection.state_changed", {"old": old_state, "new": state})
+
     async def _on_ice_trickle(self, event):
         """Handle ICE trickle from SFU."""
         logger.debug(f"Received ICE trickle for peer type {event.peer_type}")
@@ -124,6 +124,75 @@ async def _on_ice_trickle(self, event):
         except Exception as e:
             logger.debug(f"Error handling ICE trickle: {e}")
 
+    async def _on_subscriber_offer(self, event):
+        """Handle subscriber offer from SFU."""
+        logger.info(f"Received subscriber offer: ice_restart={event.ice_restart}")
+
+        try:
+            # Ensure we have a subscriber peer connection
+            if not self.subscriber_pc:
+                await self._peer_manager.setup_subscriber()
+
+            # Parse SDP to extract track-to-stream mapping
+            self._extract_track_stream_mapping(event.sdp)
+
+            # Handle ICE restart if needed
+            if event.ice_restart:
+                logger.info("Restarting ICE for subscriber")
+                await self.subscriber_pc.restartIce()
+
+            # Set remote description with the SFU's offer
+            remote_description = aiortc.RTCSessionDescription(
+                type="offer", sdp=event.sdp
+            )
+            await self.subscriber_pc.setRemoteDescription(remote_description)
+
+            # Create and set local answer
+            answer = await self.subscriber_pc.createAnswer()
+            await self.subscriber_pc.setLocalDescription(answer)
+
+            # Send answer back to SFU
+            response = await self.twirp_signaling_client.SendAnswer(
+                ctx=self.twirp_context,
+                request=signal_pb2.SendAnswerRequest(
+                    session_id=self.session_id,
+                    peer_type=models_pb2.PEER_TYPE_SUBSCRIBER,
+                    sdp=self.subscriber_pc.localDescription.sdp,
+                ),
+                server_path_prefix="",
+            )
+            logger.info(f"Sent subscriber answer: {response}")
+
+        except Exception as e:
+            logger.error(f"Error handling subscriber offer: {e}")
+            raise
+
+    def _extract_track_stream_mapping(self, sdp: str):
+        """Extract track-to-stream mapping from SDP."""
+        track_mapping = {}
+
+        # Parse SDP to find track-to-stream mapping
+        # SDP format includes lines like:
+        # a=msid:<stream_id> <track_id>
+        # a=mid:<media_id>
+        for line in sdp.split("\n"):
+            line = line.strip()
+            if line.startswith("a=msid:"):
+                # Extract msid line: a=msid:<stream_id> <track_id>
+                parts = line.split(" ")
+                if len(parts) >= 3:
+                    stream_id = parts[1]
+                    track_id = parts[2]
+                    track_mapping[track_id] = stream_id
+                    logger.debug(f"Extracted track mapping: {track_id} -> {stream_id}")
+
+        # Set the mapping in participants state
+        if track_mapping:
+            logger.info(f"Setting track stream mapping: {track_mapping}")
+            self.participants_state.set_track_stream_mapping(track_mapping)
+        else:
+            logger.warning("No track-to-stream mapping found in SDP")
+
     async def _connect_internal(
         self,
         region: Optional[str] = None,
@@ -139,10 +208,10 @@ async def _connect_internal(
             ws_url: Optional WebSocket URL to connect to
             token: Optional authentication token
             session_id: Optional session ID
-            
+
         Raises:
             SfuConnectionError: If connection fails
-        """        
+        """
         self.connection_state = ConnectionState.JOINING
 
         # Step 1: Determine region
@@ -158,16 +227,21 @@ async def _connect_internal(
         # Step 2: Join call via coordinator
         if not (ws_url or token):
             join_response = await join_call(
-                self.call, self.user_id, location, self.create, self.local_sfu, **self.kwargs
+                self.call,
+                self.user_id,
+                location,
+                self.create,
+                self.local_sfu,
+                **self.kwargs,
             )
             ws_url = join_response.data.credentials.server.ws_endpoint
             token = join_response.data.credentials.token
             self.join_response = join_response
             logger.debug(f"coordinator join response: {join_response.data}")
-        
+
         # Use provided session_id or current one
         current_session_id = session_id or self.session_id
-        
+
         # Step 3: Connect to WebSocket
         try:
             self._ws_client, sfu_event = await connect_websocket(
@@ -180,6 +254,17 @@ async def _connect_internal(
             self._ws_client.on_wildcard("*", _log_event)
             self._ws_client.on_event("ice_trickle", self._on_ice_trickle)
 
+            # Connect track subscription events to subscription manager
+            self._ws_client.on_event(
+                "track_published", self._subscription_manager.handle_track_published
+            )
+            self._ws_client.on_event(
+                "track_unpublished", self._subscription_manager.handle_track_unpublished
+            )
+
+            # Connect subscriber offer event to handle SDP negotiation
+            self._ws_client.on_event("subscriber_offer", self._on_subscriber_offer)
+
             if hasattr(sfu_event, "join_response"):
                 logger.debug(f"sfu join response: {sfu_event.join_response}")
                 # Populate participants state with existing participants
@@ -193,7 +278,7 @@ async def _connect_internal(
                     )
             else:
                 logger.warning(f"No join response from WebSocket: {sfu_event}")
-                    
+
             logger.debug(f"WebSocket connected successfully to {ws_url}")
         except Exception as e:
             logger.error(f"Failed to connect WebSocket to {ws_url}: {e}")
@@ -204,15 +289,21 @@ async def _connect_internal(
         self.twirp_signaling_client = SignalClient(address=twirp_server_url)
         self.twirp_context = Context(headers={"authorization": token})
 
-        # Step 5: Create coordinator websocket
-        user_token = self.call.client.stream.create_token(user_id=self.user_id)
-        self._coordinator_ws_client = StreamAPIWS(
-            api_key=self.call.client.stream.api_key,
-            token=user_token,
-            user_details={"id": self.user_id},
-        )
-        self._coordinator_ws_client.on_wildcard("*", _log_event)
-        await self._coordinator_ws_client.connect()
+        # Step 5: Create coordinator websocket (temporarily disabled to test)
+        # user_token = self.call.client.stream.create_token(user_id=self.user_id)
+        # self._coordinator_ws_client = StreamAPIWS(
+        #     api_key=self.call.client.stream.api_key,
+        #     token=user_token,
+        #     user_details={"id": self.user_id},
+        #     healthcheck_interval=15.0,  # Send heartbeat every 15 seconds instead of 25
+        #     healthcheck_timeout=20.0,   # Expect server messages within 20 seconds instead of 30
+        # )
+        # self._coordinator_ws_client.on_wildcard("*", _log_event)
+        # await self._coordinator_ws_client.connect()
+        self._coordinator_ws_client = None  # Temporarily disable coordinator connection
+
+        # Step 6: Setup subscriber peer connection to receive incoming tracks
+        await self._peer_manager.setup_subscriber()
 
         # Mark as connected
         self.running = True
@@ -255,6 +346,9 @@ async def leave(self):
         if self._ws_client:
             await self._ws_client.close()
             self._ws_client = None
+        if self._coordinator_ws_client:
+            await self._coordinator_ws_client.disconnect()
+            self._coordinator_ws_client = None
 
         self.connection_state = ConnectionState.LEFT
 
@@ -264,7 +358,7 @@ async def __aenter__(self):
         """Async context manager entry."""
         # Register network event handlers
         self._network_monitor.register_event_handlers()
-        
+
         # Connect with retry
         await self.connect()
 
@@ -287,10 +381,14 @@ async def addTrack(self, track, track_info=None):
         else:
             await self.add_tracks(audio=track)
 
-    async def start_recording(self, recording_types, user_ids=None, output_dir="recordings"):
+    async def start_recording(
+        self, recording_types, user_ids=None, output_dir="recordings"
+    ):
         """Start recording."""
         logger.info("Starting recording")
-        await self._recording_manager.start_recording(recording_types, user_ids, output_dir)
+        await self._recording_manager.start_recording(
+            recording_types, user_ids, output_dir
+        )
 
     async def stop_recording(self, recording_types=None, user_ids=None):
         """Stop recording."""
@@ -306,15 +404,19 @@ def get_recording_status(self) -> dict:
         """Get current recording status."""
         return self._recording_manager.get_recording_status()
 
-    async def subscribe_to_track(self, track_id: str, config: Optional[SubscriptionConfig] = None):
+    async def subscribe_to_track(
+        self, track_id: str, config: Optional[SubscriptionConfig] = None
+    ):
         """Subscribe to a specific track."""
         await self._subscription_manager.subscribe_to_track(track_id, config)
 
     async def unsubscribe_from_track(self, track_id: str):
         """Unsubscribe from a specific track."""
         await self._subscription_manager.unsubscribe_from_track(track_id)
 
-    async def update_track_subscription(self, track_id: str, config: SubscriptionConfig):
+    async def update_track_subscription(
+        self, track_id: str, config: SubscriptionConfig
+    ):
         """Update subscription configuration for a track."""
         await self._subscription_manager.update_track_subscription(track_id, config)
 
@@ -410,7 +512,9 @@ def subscriber_negotiation_lock(self):
     # Internal cleanup & restoration helpers referenced by other modules
     # ------------------------------------------------------------------
 
-    async def _cleanup_connections(self, ws_client=None, publisher_pc=None, subscriber_pc=None):
+    async def _cleanup_connections(
+        self, ws_client=None, publisher_pc=None, subscriber_pc=None
+    ):
         """Close provided connections safely; used by ReconnectionManager."""
         try:
             # Close peer connections (async)
@@ -438,4 +542,3 @@ async def _restore_published_tracks(self):
             await self._peer_manager.restore_published_tracks()
         except Exception as e:
             logger.error("Failed to restore published tracks", exc_info=e)
-

getstream/video/rtc/coordinator/ws.py

@@ -94,9 +94,6 @@ def _build_auth_payload(self) -> dict:
         """
         Build the authentication payload to send after connection.
 
-        For initial connections, includes user_details if provided.
-        For reconnections, user_details are excluded.
-
         Returns:
             Authentication payload as a dictionary
         """
@@ -105,8 +102,8 @@ def _build_auth_payload(self) -> dict:
             "products": ["video"],
         }
 
-        # Only include user_details on initial connection, not on reconnects
-        if self._initial_connection and self.user_details:
+        # Include user_details if available (both for initial connection and reconnections)
+        if self.user_details:
             payload["user_details"] = self.user_details
 
         return payload

getstream/video/rtc/participants.py

@@ -23,10 +23,21 @@ def __init__(self):
     # ------------------------------------------------------------------
 
     def get_user_from_track_id(self, track_id: str) -> Optional[models_pb2.Participant]:
+        # Track IDs have format: participant_id:track_type:...
+        # We can extract the participant prefix directly from the track ID
+        if ":" in track_id:
+            # Extract the participant prefix from the track ID
+            prefix = track_id.split(":")[0]
+            user = self._participant_by_prefix.get(prefix)
+            if user:
+                return user
+
+        # Fallback to the old mapping approach if it exists
         stream_id = self._track_stream_mapping.get(track_id)
         if stream_id:
             prefix = stream_id.split(":")[0]
             return self._participant_by_prefix.get(prefix)
+
         return None
 
     def get_stream_id_from_track_id(self, track_id: str) -> Optional[str]:
@@ -57,4 +68,4 @@ async def _on_participant_joined(self, event: events_pb2.ParticipantJoined):
 
     async def _on_participant_left(self, event: events_pb2.ParticipantLeft):
         self.remove_participant(event.participant)
-        self.emit("participant_left", event.participant) 
+        self.emit("participant_left", event.participant)