Fix track mapping when single peer connectionis used (#1696)

Co-authored-by: boks1971 <[email protected]>

Author: lukasIO

.changeset/thick-wombats-battle.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Fix track mapping when single peer connectionis used

package.json

@@ -56,7 +56,7 @@
   },
   "dependencies": {
     "@livekit/mutex": "1.1.1",
-    "@livekit/protocol": "1.42.0",
+    "@livekit/protocol": "1.42.2",
     "events": "^3.3.0",
     "jose": "^6.1.0",
     "loglevel": "^1.9.2",

pnpm-lock.yaml

@@ -124,9 +124,17 @@ export class SignalClient {
 
   onClose?: (reason: string) => void;
 
-  onAnswer?: (sd: RTCSessionDescriptionInit, offerId: number) => void;
-
-  onOffer?: (sd: RTCSessionDescriptionInit, offerId: number) => void;
+  onAnswer?: (
+    sd: RTCSessionDescriptionInit,
+    offerId: number,
+    midToTrackId: { [key: string]: string },
+  ) => void;
+
+  onOffer?: (
+    sd: RTCSessionDescriptionInit,
+    offerId: number,
+    midToTrackId: { [key: string]: string },
+  ) => void;
 
   // when a new ICE candidate is made available
   onTrickle?: (sd: RTCIceCandidateInit, target: SignalTarget) => void;
@@ -715,12 +723,12 @@ export class SignalClient {
     if (msg.case === 'answer') {
       const sd = fromProtoSessionDescription(msg.value);
       if (this.onAnswer) {
-        this.onAnswer(sd, msg.value.id);
+        this.onAnswer(sd, msg.value.id, msg.value.midToTrackId);
       }
     } else if (msg.case === 'offer') {
       const sd = fromProtoSessionDescription(msg.value);
       if (this.onOffer) {
-        this.onOffer(sd, msg.value.id);
+        this.onOffer(sd, msg.value.id, msg.value.midToTrackId);
       }
     } else if (msg.case === 'trickle') {
       const candidate: RTCIceCandidateInit = JSON.parse(msg.value.candidateInit!);

src/api/SignalClient.ts

@@ -256,6 +256,16 @@ export class PCTransportManager {
     return this.publisher.addTransceiverOfKind(kind, transceiverInit);
   }
 
+  getMidForReceiver(receiver: RTCRtpReceiver): string | null | undefined {
+    const transceivers = this.subscriber
+      ? this.subscriber.getTransceivers()
+      : this.publisher.getTransceivers();
+    const matchingTransceiver = transceivers.find(
+      (transceiver) => transceiver.receiver === receiver,
+    );
+    return matchingTransceiver?.mid;
+  }
+
   addPublisherTrack(track: MediaStreamTrack) {
     return this.publisher.addTrack(track);
   }

src/room/PCTransportManager.ts

@@ -202,6 +202,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   private reliableReceivedState: TTLMap<string, number> = new TTLMap(reliabeReceiveStateTTL);
 
+  private midToTrackId: { [key: string]: string } = {};
+
   constructor(private options: InternalRoomOptions) {
     super();
     this.log = getLogger(options.loggerName ?? LoggerNames.Engine);
@@ -499,15 +501,17 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   private setupSignalClientCallbacks() {
     // configure signaling client
-    this.client.onAnswer = async (sd, offerId) => {
+    this.client.onAnswer = async (sd, offerId, midToTrackId) => {
       if (!this.pcManager) {
         return;
       }
       this.log.debug('received server answer', {
         ...this.logContext,
         RTCSdpType: sd.type,
         sdp: sd.sdp,
+        midToTrackId,
       });
+      this.midToTrackId = midToTrackId;
       await this.pcManager.setPublisherAnswer(sd, offerId);
     };
 
@@ -521,11 +525,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     };
 
     // when server creates an offer for the client
-    this.client.onOffer = async (sd, offerId) => {
+    this.client.onOffer = async (sd, offerId, midToTrackId) => {
       this.latestRemoteOfferId = offerId;
       if (!this.pcManager) {
         return;
       }
+      this.midToTrackId = midToTrackId;
       const answer = await this.pcManager.createSubscriberAnswerFromOffer(sd, offerId);
       if (answer) {
         this.client.sendAnswer(answer, offerId);
@@ -1629,6 +1634,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       window.removeEventListener('online', this.handleBrowserOnLine);
     }
   }
+
+  getTrackIdForReceiver(receiver: RTCRtpReceiver): string | undefined {
+    const mid = this.pcManager?.getMidForReceiver(receiver);
+    if (mid) {
+      const match = Object.entries(this.midToTrackId).find(([key]) => key === mid);
+      if (match) {
+        return match[1];
+      }
+    }
+  }
 }
 
 class SignalReconnectError extends Error {}

src/room/RTCEngine.ts

@@ -1370,6 +1370,11 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     // We'll defer these events until when the room is connected or eventually disconnected.
     if (this.state === ConnectionState.Connecting || this.state === ConnectionState.Reconnecting) {
       const reconnectedHandler = () => {
+        this.log.debug('deferring on track for later', {
+          mediaTrackId: mediaTrack.id,
+          mediaStreamId: stream.id,
+          tracksInStream: stream.getTracks().map((track) => track.id),
+        });
         this.onTrackAdded(mediaTrack, stream, receiver);
         cleanup();
       };
@@ -1416,6 +1421,28 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       return;
     }
 
+    // in single peer connection case, the trackID is locally generated,
+    // not the TR_ prefixed one generated by the server,
+    // use `mid` to find the appropriate track.
+    if (!trackId.startsWith('TR')) {
+      const id = this.engine.getTrackIdForReceiver(receiver);
+      if (!id) {
+        this.log.error(
+          `Tried to add a track whose 'sid' could not be found for a participant, that's not present. Sid: ${participantSid}`,
+          this.logContext,
+        );
+        return;
+      }
+
+      trackId = id;
+    }
+    if (!trackId.startsWith('TR')) {
+      this.log.warn(
+        `Tried to add a track whose 'sid' could not be determined for a participant, that's not present. Sid: ${participantSid}, streamId: ${streamId}, trackId: ${trackId}`,
+        { ...this.logContext, rpID: participantSid, streamId, trackId },
+      );
+    }
+
     let adaptiveStreamSettings: AdaptiveStreamSettings | undefined;
     if (this.options.adaptiveStream) {
       if (typeof this.options.adaptiveStream === 'object') {
@@ -2541,6 +2568,13 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     if (event !== RoomEvent.ActiveSpeakersChanged && event !== RoomEvent.TranscriptionReceived) {
       // only extract logContext from arguments in order to avoid logging the whole object tree
       const minimizedArgs = mapArgs(args).filter((arg: unknown) => arg !== undefined);
+      if (event === RoomEvent.TrackSubscribed || event === RoomEvent.TrackUnsubscribed) {
+        this.log.trace(`subscribe trace: ${event}`, {
+          ...this.logContext,
+          event,
+          args: minimizedArgs,
+        });
+      }
       this.log.debug(`room event ${event}`, { ...this.logContext, event, args: minimizedArgs });
     }
     return super.emit(event, ...args);