chore: fix merge conflict issues

Author: vivekd22

examples/record_audio/main.py

@@ -91,18 +91,23 @@ async def record(connection, recording_type, output_dir, user_id_filter=None):
     print(f"   • Output Directory: {final_status['output_directory']}")
 
 
-async def record_composite(call, recorder_connection, bot_connections, players):
+async def record_composite(recorder_connection, bot_connections):
     """Record composite audio from all participants."""
-    await record(recorder_connection, RecordingType.COMPOSITE, "recordings/composite")
+    await record(
+        recorder_connection,
+        RecordingType.COMPOSITE,
+        "recordings/composite",
+        user_id_filter=bot_connections,
+    )
 
 
-async def record_tracks(call, recorder_connection, bot_connections, players):
+async def record_tracks(recorder_connection, bot_connections):
     """Record individual tracks from each participant."""
     await record(
         recorder_connection,
         RecordingType.TRACK,
         "recordings/tracks",
-        user_id_filter=bot_connections[0].user_id,  # Record only the first bot's audio
+        user_id_filter=bot_connections,
     )
 
 
@@ -189,17 +194,13 @@ async def main():
                 # Run the selected recording type
                 if args.type == "composite":
                     await record_composite(
-                        call,
                         recorder_connection,
-                        [bot1_connection, bot2_connection, bot3_connection],
-                        players,
+                        [bot1_connection.user_id],
                     )
                 else:  # track recording
                     await record_tracks(
-                        call,
                         recorder_connection,
-                        [bot1_connection, bot2_connection, bot3_connection],
-                        players,
+                        [bot1_connection.user_id],
                     )
 
                 for connection in [bot1_connection, bot2_connection, bot3_connection]:

getstream/video/rtc/connection_manager.py

@@ -9,9 +9,11 @@
 from twirp.context import Context
 
 from getstream.utils import StreamAsyncIOEventEmitter
+from getstream.video.rtc.coordinator.ws import StreamAPIWS
+from getstream.video.rtc.pb.stream.video.sfu.event import events_pb2
 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.rtc.twirp_client_wrapper import SfuRpcError, SignalClient
 
 from getstream.video.call import Call
 from getstream.video.rtc.connection_utils import (
@@ -21,6 +23,10 @@
     connect_websocket,
     join_call,
 )
+from getstream.video.rtc.track_util import (
+    fix_sdp_msid_semantic,
+    parse_track_stream_mapping,
+)
 from getstream.video.rtc.network_monitor import NetworkMonitor
 from getstream.video.rtc.recording import RecordingManager
 from getstream.video.rtc.participants import ParticipantsState
@@ -72,6 +78,7 @@ def __init__(
         self._recording_manager: RecordingManager = RecordingManager()
         self._network_monitor: NetworkMonitor = NetworkMonitor(self)
         self._reconnector: ReconnectionManager = ReconnectionManager(self)
+        logger.info(f"VIVEK subscription_config: {subscription_config}")
         self._subscription_manager: SubscriptionManager = SubscriptionManager(
             self, subscription_config
         )
@@ -124,48 +131,57 @@ 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)
+    async def _on_subscriber_offer(self, event: events_pb2.SubscriberOffer):
+        logger.info("Subscriber offer received")
 
-            # Handle ICE restart if needed
-            if event.ice_restart:
-                logger.info("Restarting ICE for subscriber")
-                await self.subscriber_pc.restartIce()
+        await self.subscriber_negotiation_lock.acquire()
 
-            # Set remote description with the SFU's offer
+        try:
+            # Fix any invalid msid-semantic format in the SDP
+            fixed_sdp = fix_sdp_msid_semantic(event.sdp)
+            # Parse SDP to create track_id to stream_id mapping
+            self.participants_state.set_track_stream_mapping(
+                parse_track_stream_mapping(fixed_sdp)
+            )
+            # The SDP offer from the SFU might already contain candidates (trickled)
+            # or have a different structure. We set it as the remote description.
+            # The aiortc library handles merging and interpretation.
             remote_description = aiortc.RTCSessionDescription(
-                type="offer", sdp=event.sdp
+                type="offer", sdp=fixed_sdp
             )
+            logger.debug(f"""Setting remote description with SDP:
+            {remote_description.sdp}""")
             await self.subscriber_pc.setRemoteDescription(remote_description)
 
-            # Create and set local answer
+            # Create the answer based on the remote offer (which includes our candidates)
             answer = await self.subscriber_pc.createAnswer()
+            # Set the local description. aiortc will manage the SDP content.
             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"""Sending answer with local description:
+            {self.subscriber_pc.localDescription.sdp}"""
             )
-            logger.info(f"Sent subscriber answer: {response}")
 
-        except Exception as e:
-            logger.error(f"Error handling subscriber offer: {e}")
-            raise
+            try:
+                await self.twirp_signaling_client.SendAnswer(
+                    ctx=self.twirp_context,
+                    request=signal_pb2.SendAnswerRequest(
+                        peer_type=models_pb2.PEER_TYPE_SUBSCRIBER,
+                        sdp=self.subscriber_pc.localDescription.sdp,
+                        session_id=self.session_id,
+                    ),
+                    server_path_prefix="",  # Note: Our wrapper doesn't need this, underlying client handles prefix
+                )
+                logger.info("Subscriber answer sent successfully.")
+            except SfuRpcError as e:
+                logger.error(f"Failed to send subscriber answer: {e}")
+                # Decide how to handle: maybe close connection, notify user, etc.
+                # For now, just log the error.
+            except Exception as e:
+                logger.error(f"Unexpected error sending subscriber answer: {e}")
+        finally:
+            self.subscriber_negotiation_lock.release()
 
     def _extract_track_stream_mapping(self, sdp: str):
         """Extract track-to-stream mapping from SDP."""
@@ -242,6 +258,8 @@ async def _connect_internal(
         # Use provided session_id or current one
         current_session_id = session_id or self.session_id
 
+        await self._peer_manager.setup_subscriber()
+
         # Step 3: Connect to WebSocket
         try:
             self._ws_client, sfu_event = await connect_websocket(
@@ -290,20 +308,14 @@ async def _connect_internal(
         self.twirp_context = Context(headers={"authorization": token})
 
         # 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()
+        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()
 
         # Mark as connected
         self.running = True

getstream/video/rtc/coordinator/ws.py

@@ -41,7 +41,7 @@ def __init__(
         user_details: Optional[dict] = None,
         *,
         uri: str = DEFAULT_WS_URI,
-        healthcheck_interval: float = 25.0,
+        healthcheck_interval: float = 15.0,
         healthcheck_timeout: float = 30.0,
         max_retries: int = 5,
         backoff_base: float = 1.0,

getstream/video/rtc/pc.py

@@ -118,7 +118,7 @@ def __init__(
 
         @self.on("track")
         async def on_track(track: aiortc.mediastreams.MediaStreamTrack):
-            logger.info(f"VIVEK Track received: {track.id} : {track.kind}")
+            logger.info(f"Track received: {track.id} : {track.kind}")
 
             # Try to get user from track ID first (original method)
             user = self.connection.participants_state.get_user_from_track_id(track.id)

getstream/video/rtc/recording.py

@@ -702,6 +702,10 @@ def __init__(self, config: Optional[RecordingConfig] = None):
 
         self.config = config or RecordingConfig()
 
+        # Tracks that were received before recording started.
+        # Mapping of track_id -> (user_id, track)
+        self._pending_tracks: Dict[str, tuple[str, MediaStreamTrack]] = {}
+
         # Recording state
         self._recording_types: Set[RecordingType] = set()
         self._target_user_ids: Optional[Set[str]] = None
@@ -752,6 +756,9 @@ async def start_recording(
         # Set up event forwarding for new recorders
         self._setup_recorder_events()
 
+        # Process any tracks that were received before recording was enabled.
+        await self._process_pending_tracks()
+
         self.emit('recording_started', {
             'recording_types': [rt.value for rt in recording_types],
             'user_ids': user_ids,
@@ -803,6 +810,12 @@ async def on_user_left(self, user_id: str):
         if self._composite_audio_recorder:
             self._composite_audio_recorder.remove_user_track(user_id)
 
+        # Remove from pending tracks
+        for track_id in list(self._pending_tracks.keys()):
+            pending_user, _ = self._pending_tracks[track_id]
+            if pending_user == user_id:
+                del self._pending_tracks[track_id]
+
         # Stop track recording (both audio and video)
         await self._stop_user_recording(user_id)
         logger.info(f"Stopped recording for user {user_id} who left the call")
@@ -814,8 +827,13 @@ async def on_track_removed(self, user_id: str, track_type: str):
             await self._stop_user_recording_by_key(recorder_key, track_type)
             logger.info(f"Stopped {track_type} recording for user {user_id} whose {track_type} track was removed")
 
+            # Remove from pending
+            for track_id in list(self._pending_tracks.keys()):
+                pending_user, pending_track = self._pending_tracks[track_id]
+                if pending_user == user_id and pending_track.kind == track_type:
+                    del self._pending_tracks[track_id]
+
     async def on_track_received(self, track, user):
-        logger.info(f"VIVEK on_track_received: {track.kind} track for user {user}")
         """Handle new track received from track event."""
         if not track or track.kind not in ["audio", "video"]:
             return
@@ -838,6 +856,11 @@ async def on_track_received(self, track, user):
         if RecordingType.TRACK in self._recording_types:
             await self._start_user_track_recording(user_id, track)
 
+        # Cache track so that we can start recording later if recording is not
+        # yet enabled.
+        if RecordingType.TRACK not in self._recording_types:
+            self._pending_tracks[track.id] = (user_id, track)
+
     async def _start_user_track_recording(self, user_id: str, track: MediaStreamTrack):
         """Start recording a user track using MediaRecorder."""
         track_type = TrackType.AUDIO if track.kind == "audio" else TrackType.VIDEO
@@ -1166,3 +1189,31 @@ async def cleanup(self):
             self._composite_audio_recorder = None
 
         logger.info("RecordingManager cleanup completed")
+
+    async def _process_pending_tracks(self):
+        """Start recording for any tracks that were received before recording was enabled."""
+        if not self._pending_tracks:
+            return
+
+        for track_id, (user_id, track) in list(self._pending_tracks.items()):
+            try:
+                # Filter by requested user_ids if provided
+                if self._target_user_ids and user_id not in self._target_user_ids:
+                    continue
+
+                # Start individual track recording if requested
+                if RecordingType.TRACK in self._recording_types:
+                    await self._start_user_track_recording(user_id, track)
+
+                # Add to composite recorder if requested and audio track
+                if (
+                    RecordingType.COMPOSITE in self._recording_types
+                    and track.kind == "audio"
+                    and self._composite_audio_recorder is not None
+                ):
+                    self._composite_audio_recorder.add_user_track(user_id, track)
+
+                # Remove from pending once handled
+                del self._pending_tracks[track_id]
+            except Exception as e:
+                logger.warning(f"Error processing pending track {track_id}: {e}")