Only reject connection promise if it was triggered by a call to connect() (#359)

* Only reject connection promise if it was triggered by a call to connect()

* changeset

* immediately return promise from connect function, clean up failed connection attempts

* address comments

* only reject promise in disconnect if it's a reconnection

Author: lukasIO

.changeset/breezy-chefs-know.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Only reject connection promise if it was triggered by a call to connect()

src/room/Room.ts

@@ -99,8 +99,12 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
   /** used for aborting pending connections to a LiveKit server */
   private abortController?: AbortController;
 
+  /** future holding client initiated connection attempt */
   private connectFuture?: Future<void>;
 
+  /** future holding sdk initiated reconnection attempt */
+  private reconnectFuture?: Future<void>;
+
   /**
    * Creates a new Room, the primary construct for a LiveKit session.
    * @param options
@@ -156,12 +160,17 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       .on(EngineEvent.ActiveSpeakersUpdate, this.handleActiveSpeakersUpdate)
       .on(EngineEvent.DataPacketReceived, this.handleDataPacket)
       .on(EngineEvent.Resuming, () => {
+        if (!this.reconnectFuture) {
+          this.reconnectFuture = new Future();
+        }
         if (this.setAndEmitConnectionState(ConnectionState.Reconnecting)) {
           this.emit(RoomEvent.Reconnecting);
         }
       })
       .on(EngineEvent.Resumed, () => {
         this.setAndEmitConnectionState(ConnectionState.Connected);
+        this.reconnectFuture?.resolve();
+        this.reconnectFuture = undefined;
         this.emit(RoomEvent.Reconnected);
         this.updateSubscriptions();
       })
@@ -193,157 +202,159 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     return DeviceManager.getInstance().getDevices(kind, requestPermissions);
   }
 
-  connect = async (url: string, token: string, opts?: RoomConnectOptions): Promise<void> => {
+  connect = (url: string, token: string, opts?: RoomConnectOptions): Promise<void> => {
     if (this.state === ConnectionState.Connected) {
       // when the state is reconnecting or connected, this function returns immediately
       log.warn(`already connected to room ${this.name}`);
-      return;
+      return Promise.resolve();
     }
 
     if (this.connectFuture) {
       return this.connectFuture.promise;
+    } else if (this.reconnectFuture) {
+      this.connectFuture = this.reconnectFuture;
+      return this.connectFuture.promise;
     }
-    this.setAndEmitConnectionState(ConnectionState.Connecting);
+    const connectPromise = new Promise<void>(async (resolve, reject) => {
+      this.setAndEmitConnectionState(ConnectionState.Connecting);
+      if (!this.abortController || this.abortController.signal.aborted) {
+        this.abortController = new AbortController();
+      }
 
-    if (!this.abortController || this.abortController.signal.aborted) {
-      this.abortController = new AbortController();
-    }
+      // recreate engine if previously disconnected
+      this.createEngine();
 
-    // recreate engine if previously disconnected
-    this.createEngine();
+      this.acquireAudioContext();
 
-    this.acquireAudioContext();
+      if (opts?.rtcConfig) {
+        this.engine.rtcConfig = opts.rtcConfig;
+      }
 
-    if (opts?.rtcConfig) {
-      this.engine.rtcConfig = opts.rtcConfig;
-    }
+      this.connOptions = opts;
+
+      try {
+        const joinResponse = await this.engine.join(
+          url,
+          token,
+          {
+            autoSubscribe: opts?.autoSubscribe,
+            publishOnly: opts?.publishOnly,
+            adaptiveStream:
+              typeof this.options?.adaptiveStream === 'object'
+                ? true
+                : this.options?.adaptiveStream,
+          },
+          this.abortController.signal,
+        );
+        log.debug(
+          `connected to Livekit Server version: ${joinResponse.serverVersion}, region: ${joinResponse.serverRegion}`,
+        );
 
-    this.connOptions = opts;
+        if (!joinResponse.serverVersion) {
+          throw new UnsupportedServer('unknown server version');
+        }
 
-    try {
-      const joinResponse = await this.engine.join(
-        url,
-        token,
-        {
-          autoSubscribe: opts?.autoSubscribe,
-          publishOnly: opts?.publishOnly,
-          adaptiveStream:
-            typeof this.options?.adaptiveStream === 'object' ? true : this.options?.adaptiveStream,
-        },
-        this.abortController.signal,
-      );
-      log.debug(
-        `connected to Livekit Server version: ${joinResponse.serverVersion}, region: ${joinResponse.serverRegion}`,
-      );
+        if (joinResponse.serverVersion === '0.15.1' && this.options.dynacast) {
+          log.debug('disabling dynacast due to server version');
+          // dynacast has a bug in 0.15.1, so we cannot use it then
+          this.options.dynacast = false;
+        }
 
-      if (!joinResponse.serverVersion) {
-        throw new UnsupportedServer('unknown server version');
-      }
+        const pi = joinResponse.participant!;
+
+        this.localParticipant.sid = pi.sid;
+        this.localParticipant.identity = pi.identity;
+
+        this.localParticipant.updateInfo(pi);
+        // forward metadata changed for the local participant
+        this.localParticipant
+          .on(ParticipantEvent.ParticipantMetadataChanged, this.onLocalParticipantMetadataChanged)
+          .on(ParticipantEvent.TrackMuted, this.onLocalTrackMuted)
+          .on(ParticipantEvent.TrackUnmuted, this.onLocalTrackUnmuted)
+          .on(ParticipantEvent.LocalTrackPublished, this.onLocalTrackPublished)
+          .on(ParticipantEvent.LocalTrackUnpublished, this.onLocalTrackUnpublished)
+          .on(ParticipantEvent.ConnectionQualityChanged, this.onLocalConnectionQualityChanged)
+          .on(ParticipantEvent.MediaDevicesError, this.onMediaDevicesError)
+          .on(
+            ParticipantEvent.ParticipantPermissionsChanged,
+            this.onLocalParticipantPermissionsChanged,
+          );
 
-      if (joinResponse.serverVersion === '0.15.1' && this.options.dynacast) {
-        log.debug('disabling dynacast due to server version');
-        // dynacast has a bug in 0.15.1, so we cannot use it then
-        this.options.dynacast = false;
-      }
+        // populate remote participants, these should not trigger new events
+        joinResponse.otherParticipants.forEach((info) => {
+          if (
+            info.sid !== this.localParticipant.sid &&
+            info.identity !== this.localParticipant.identity
+          ) {
+            this.getOrCreateParticipant(info.sid, info);
+          } else {
+            log.warn('received info to create local participant as remote participant', {
+              info,
+              localParticipant: this.localParticipant,
+            });
+          }
+        });
 
-      const pi = joinResponse.participant!;
-
-      this.localParticipant.sid = pi.sid;
-      this.localParticipant.identity = pi.identity;
-
-      this.localParticipant.updateInfo(pi);
-      // forward metadata changed for the local participant
-      this.localParticipant
-        .on(ParticipantEvent.ParticipantMetadataChanged, this.onLocalParticipantMetadataChanged)
-        .on(ParticipantEvent.TrackMuted, this.onLocalTrackMuted)
-        .on(ParticipantEvent.TrackUnmuted, this.onLocalTrackUnmuted)
-        .on(ParticipantEvent.LocalTrackPublished, this.onLocalTrackPublished)
-        .on(ParticipantEvent.LocalTrackUnpublished, this.onLocalTrackUnpublished)
-        .on(ParticipantEvent.ConnectionQualityChanged, this.onLocalConnectionQualityChanged)
-        .on(ParticipantEvent.MediaDevicesError, this.onMediaDevicesError)
-        .on(
-          ParticipantEvent.ParticipantPermissionsChanged,
-          this.onLocalParticipantPermissionsChanged,
-        );
+        this.name = joinResponse.room!.name;
+        this.sid = joinResponse.room!.sid;
+        this.metadata = joinResponse.room!.metadata;
+        this.emit(RoomEvent.SignalConnected);
+      } catch (err) {
+        this.recreateEngine();
+        this.handleDisconnect(this.options.stopLocalTrackOnUnpublish);
+        reject(new ConnectionError('could not establish signal connection'));
+      }
 
-      // populate remote participants, these should not trigger new events
-      joinResponse.otherParticipants.forEach((info) => {
-        if (
-          info.sid !== this.localParticipant.sid &&
-          info.identity !== this.localParticipant.identity
-        ) {
-          this.getOrCreateParticipant(info.sid, info);
-        } else {
-          log.warn('received info to create local participant as remote participant', {
-            info,
-            localParticipant: this.localParticipant,
-          });
+      // don't return until ICE connected
+      const connectTimeout = setTimeout(() => {
+        // timeout
+        this.recreateEngine();
+        this.handleDisconnect(this.options.stopLocalTrackOnUnpublish);
+        reject(new ConnectionError('could not connect PeerConnection after timeout'));
+      }, maxICEConnectTimeout);
+      const abortHandler = () => {
+        log.warn('closing engine');
+        clearTimeout(connectTimeout);
+        this.recreateEngine();
+        this.handleDisconnect(this.options.stopLocalTrackOnUnpublish);
+        reject(new ConnectionError('room connection has been cancelled'));
+      };
+      if (this.abortController?.signal.aborted) {
+        abortHandler();
+      }
+      this.abortController?.signal.addEventListener('abort', abortHandler);
+
+      this.engine.once(EngineEvent.Connected, () => {
+        clearTimeout(connectTimeout);
+        this.abortController?.signal.removeEventListener('abort', abortHandler);
+        // also hook unload event
+        if (isWeb()) {
+          window.addEventListener('beforeunload', this.onBeforeUnload);
+          navigator.mediaDevices?.addEventListener('devicechange', this.handleDeviceChange);
         }
+        this.setAndEmitConnectionState(ConnectionState.Connected);
+        resolve();
       });
-
-      this.name = joinResponse.room!.name;
-      this.sid = joinResponse.room!.sid;
-      this.metadata = joinResponse.room!.metadata;
-      this.emit(RoomEvent.SignalConnected);
-    } catch (err) {
-      this.recreateEngine();
-      this.setAndEmitConnectionState(
-        ConnectionState.Disconnected,
-        new ConnectionError('could not establish signal connection'),
-      );
-      throw err;
-    }
-
-    // don't return until ICE connected
-    const connectTimeout = setTimeout(() => {
-      // timeout
-      this.recreateEngine();
-      this.setAndEmitConnectionState(
-        ConnectionState.Disconnected,
-        new ConnectionError('could not connect PeerConnection after timeout'),
-      );
-    }, maxICEConnectTimeout);
-    const abortHandler = () => {
-      log.warn('closing engine');
-      clearTimeout(connectTimeout);
-      this.recreateEngine();
-      this.setAndEmitConnectionState(
-        ConnectionState.Disconnected,
-        new ConnectionError('room connection has been cancelled'),
-      );
-    };
-    if (this.abortController?.signal.aborted) {
-      abortHandler();
-    }
-    this.abortController?.signal.addEventListener('abort', abortHandler);
-
-    this.engine.once(EngineEvent.Connected, () => {
-      clearTimeout(connectTimeout);
-      this.abortController?.signal.removeEventListener('abort', abortHandler);
-      // also hook unload event
-      if (isWeb()) {
-        window.addEventListener('beforeunload', this.onBeforeUnload);
-        navigator.mediaDevices?.addEventListener('devicechange', this.handleDeviceChange);
-      }
-      this.setAndEmitConnectionState(ConnectionState.Connected);
     });
+    this.connectFuture = new Future(connectPromise);
 
-    if (this.connectFuture) {
-      /** @ts-ignore */
-      return this.connectFuture.promise;
-    }
+    this.connectFuture.promise.finally(() => (this.connectFuture = undefined));
+
+    return this.connectFuture.promise;
   };
 
   /**
    * disconnects the room, emits [[RoomEvent.Disconnected]]
    */
   disconnect = async (stopTracks = true) => {
     log.info('disconnect from room', { identity: this.localParticipant.identity });
-    if (this.state === ConnectionState.Connecting) {
+    if (this.state === ConnectionState.Connecting || this.state === ConnectionState.Reconnecting) {
       // try aborting pending connection attempt
       log.warn('abort connection attempt');
       this.abortController?.abort();
-      return;
+      // in case the abort controller didn't manage to cancel the connection attempt, reject the connect promise explicitly
+      this.connectFuture?.reject(new ConnectionError('Client initiated disconnect'));
+      this.connectFuture = undefined;
     }
     // send leave
     if (this.engine?.client.isConnected) {
@@ -353,7 +364,6 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     if (this.engine) {
       this.engine.close();
     }
-
     this.handleDisconnect(stopTracks, DisconnectReason.CLIENT_INITIATED);
     /* @ts-ignore */
     this.engine = undefined;
@@ -571,6 +581,9 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
   }
 
   private handleRestarting = () => {
+    if (!this.reconnectFuture) {
+      this.reconnectFuture = new Future();
+    }
     // also unwind existing participants & existing subscriptions
     for (const p of this.participants.values()) {
       this.handleParticipantDisconnected(p.sid, p);
@@ -587,6 +600,8 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     });
     this.setAndEmitConnectionState(ConnectionState.Connected);
     this.emit(RoomEvent.Reconnected);
+    this.reconnectFuture?.resolve();
+    this.reconnectFuture = undefined;
 
     // rehydrate participants
     if (joinResponse.participant) {
@@ -630,6 +645,13 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     if (this.state === ConnectionState.Disconnected) {
       return;
     }
+    // reject potentially ongoing reconnection attempt
+    if (this.connectFuture === this.reconnectFuture) {
+      this.connectFuture?.reject(undefined);
+      this.connectFuture = undefined;
+      this.reconnectFuture = undefined;
+    }
+
     this.participants.forEach((p) => {
       p.tracks.forEach((pub) => {
         p.unpublishTrack(pub.trackSid);
@@ -1030,35 +1052,11 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     }
   }
 
-  private setAndEmitConnectionState(state: ConnectionState, error?: Error): boolean {
+  private setAndEmitConnectionState(state: ConnectionState): boolean {
     if (state === this.state) {
       // unchanged
       return false;
     }
-    switch (state) {
-      case ConnectionState.Connecting:
-      case ConnectionState.Reconnecting:
-        if (!this.connectFuture) {
-          // reuse existing connect future if possible
-          this.connectFuture = new Future<void>();
-        }
-        break;
-      case ConnectionState.Connected:
-        if (this.connectFuture) {
-          this.connectFuture.resolve();
-          this.connectFuture = undefined;
-        }
-        break;
-      case ConnectionState.Disconnected:
-        if (this.connectFuture) {
-          error ??= new Error('disconnected from Room');
-          this.connectFuture.reject(error);
-          this.connectFuture = undefined;
-        }
-        break;
-      default:
-      // nothing
-    }
     this.state = state;
     this.emit(RoomEvent.ConnectionStateChanged, this.state);
     return true;