Emit TrackUnpublished before TrackPublished (#541)

* Emit TrackUnpublished before TrackUnpublished within the same update

* changeset

* fix changeset

* LocalParticipant.unpublishTrack to be return a promise

* add negotiation timeout, use pc event obj

* add negotiationStarted event

* fix changeset

Co-authored-by: David Zhao <[email protected]>

Author: lukasIO

.changeset/curvy-years-listen.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Make `unpublishTrack` async, emit TrackUnpublished before TrackPublished within the same update

src/room/PCTransport.ts

@@ -1,3 +1,4 @@
+import EventEmitter from 'events';
 import { MediaDescription, parse, write } from 'sdp-transform';
 import { debounce } from 'ts-debounce';
 import log from '../logger';
@@ -10,8 +11,13 @@ interface TrackBitrateInfo {
   maxbr: number;
 }
 
+export const PCEvents = {
+  NegotiationStarted: 'negotiationStarted',
+  NegotiationComplete: 'negotiationComplete',
+} as const;
+
 /** @internal */
-export default class PCTransport {
+export default class PCTransport extends EventEmitter {
   pc: RTCPeerConnection;
 
   pendingCandidates: RTCIceCandidateInit[] = [];
@@ -27,6 +33,7 @@ export default class PCTransport {
   onOffer?: (offer: RTCSessionDescriptionInit) => void;
 
   constructor(config?: RTCConfiguration) {
+    super();
     this.pc = new RTCPeerConnection(config);
   }
 
@@ -56,11 +63,14 @@ export default class PCTransport {
     if (this.renegotiate) {
       this.renegotiate = false;
       this.createAndSendOffer();
+    } else if (sd.type === 'answer') {
+      this.emit(PCEvents.NegotiationComplete);
     }
   }
 
   // debounced negotiate interface
   negotiate = debounce((onError?: (e: Error) => void) => {
+    this.emit(PCEvents.NegotiationStarted);
     try {
       this.createAndSendOffer();
     } catch (e) {

src/room/RTCEngine.ts

@@ -29,7 +29,7 @@ import {
   UnexpectedConnectionState,
 } from './errors';
 import { EngineEvent } from './events';
-import PCTransport from './PCTransport';
+import PCTransport, { PCEvents } from './PCTransport';
 import type { ReconnectContext, ReconnectPolicy } from './ReconnectPolicy';
 import type LocalTrack from './track/LocalTrack';
 import type LocalVideoTrack from './track/LocalVideoTrack';
@@ -964,18 +964,36 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   }
 
   /** @internal */
-  negotiate() {
-    if (!this.publisher) {
-      return;
-    }
+  negotiate(): Promise<void> {
+    // observe signal state
+    return new Promise<void>((resolve, reject) => {
+      if (!this.publisher) {
+        reject(new NegotiationError('publisher is not defined'));
+        return;
+      }
 
-    this.hasPublished = true;
+      this.hasPublished = true;
 
-    this.publisher.negotiate((e) => {
-      if (e instanceof NegotiationError) {
-        this.fullReconnectOnNext = true;
-      }
-      this.handleDisconnect('negotiation');
+      const negotiationTimeout = setTimeout(() => {
+        reject('negotiation timed out');
+        this.handleDisconnect('negotiation');
+      }, this.peerConnectionTimeout);
+
+      this.publisher.once(PCEvents.NegotiationStarted, () => {
+        this.publisher?.once(PCEvents.NegotiationComplete, () => {
+          clearTimeout(negotiationTimeout);
+          resolve();
+        });
+      });
+
+      this.publisher.negotiate((e) => {
+        clearTimeout(negotiationTimeout);
+        reject(e);
+        if (e instanceof NegotiationError) {
+          this.fullReconnectOnNext = true;
+        }
+        this.handleDisconnect('negotiation');
+      });
     });
   }
 

src/room/participant/LocalParticipant.ts

@@ -262,7 +262,7 @@ export default class LocalParticipant extends Participant {
     } else if (track && track.track) {
       // screenshare cannot be muted, unpublish instead
       if (source === Track.Source.ScreenShare) {
-        track = this.unpublishTrack(track.track);
+        track = await this.unpublishTrack(track.track);
         const screenAudioTrack = this.getTrack(Track.Source.ScreenShareAudio);
         if (screenAudioTrack && screenAudioTrack.track) {
           this.unpublishTrack(screenAudioTrack.track);
@@ -694,10 +694,10 @@ export default class LocalParticipant extends Participant {
     log.debug(`published ${videoCodec} for track ${track.sid}`, { encodings, trackInfo: ti });
   }
 
-  unpublishTrack(
+  async unpublishTrack(
     track: LocalTrack | MediaStreamTrack,
     stopOnUnpublish?: boolean,
-  ): LocalTrackPublication | undefined {
+  ): Promise<LocalTrackPublication | undefined> {
     // look through all published tracks to find the right ones
     const publication = this.getPublicationForTrack(track);
 
@@ -744,7 +744,7 @@ export default class LocalParticipant extends Participant {
       } catch (e) {
         log.warn('failed to unpublish track', { error: e, method: 'unpublishTrack' });
       } finally {
-        this.engine.negotiate();
+        await this.engine.negotiate();
       }
     }
 
@@ -769,15 +769,33 @@ export default class LocalParticipant extends Participant {
     return publication;
   }
 
-  unpublishTracks(tracks: LocalTrack[] | MediaStreamTrack[]): LocalTrackPublication[] {
-    const publications: LocalTrackPublication[] = [];
-    tracks.forEach((track: LocalTrack | MediaStreamTrack) => {
-      const pub = this.unpublishTrack(track);
-      if (pub) {
-        publications.push(pub);
+  async unpublishTracks(
+    tracks: LocalTrack[] | MediaStreamTrack[],
+  ): Promise<LocalTrackPublication[]> {
+    const results = await Promise.all(tracks.map((track) => this.unpublishTrack(track)));
+    return results.filter(
+      (track) => track instanceof LocalTrackPublication,
+    ) as LocalTrackPublication[];
+  }
+
+  async republishAllTracks(options?: TrackPublishOptions) {
+    const localPubs: LocalTrackPublication[] = [];
+    this.tracks.forEach((pub) => {
+      if (pub.track) {
+        if (options) {
+          pub.options = { ...pub.options, ...options };
+        }
+        localPubs.push(pub);
       }
     });
-    return publications;
+
+    await Promise.all(
+      localPubs.map(async (pub) => {
+        const track = pub.track!;
+        await this.unpublishTrack(track, false);
+        await this.publishTrack(track, pub.options);
+      }),
+    );
   }
 
   /**

src/room/participant/RemoteParticipant.ts

@@ -263,11 +263,6 @@ export default class RemoteParticipant extends Participant {
       validTracks.set(ti.sid, publication);
     });
 
-    // always emit events for new publications, Room will not forward them unless it's ready
-    newTracks.forEach((publication) => {
-      this.emit(ParticipantEvent.TrackPublished, publication);
-    });
-
     // detect removed tracks
     this.tracks.forEach((publication) => {
       if (!validTracks.has(publication.trackSid)) {
@@ -278,6 +273,11 @@ export default class RemoteParticipant extends Participant {
         this.unpublishTrack(publication.trackSid, true);
       }
     });
+
+    // always emit events for new publications, Room will not forward them unless it's ready
+    newTracks.forEach((publication) => {
+      this.emit(ParticipantEvent.TrackPublished, publication);
+    });
   }
 
   /** @internal */