Add RoomEvent.Connected, fix connect future (#431)

* add connected event, fix connect future * changeset
livekit · Sep 12, 2022 · 4004426 · 4004426
1 parent e4ead3d
commit 4004426
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 18 deletions.
diff --git a/.changeset/stupid-rice-reflect.md b/.changeset/stupid-rice-reflect.md
@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Add RoomEvent.Connected, fix connectFuture rejection exception
diff --git a/src/room/Room.ts b/src/room/Room.ts
@@ -171,7 +171,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       })
       .on(EngineEvent.Resumed, () => {
         this.setAndEmitConnectionState(ConnectionState.Connected);
-        this.reconnectFuture?.resolve();
+        this.reconnectFuture?.resolve?.();
         this.reconnectFuture = undefined;
         this.emit(RoomEvent.Reconnected);
         this.updateSubscriptions();
@@ -218,7 +218,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       this.connectFuture = this.reconnectFuture;
       return this.connectFuture.promise;
     }
-    const connectPromise = new Promise<void>(async (resolve, reject) => {
+    const connectFn = async (resolve: () => void, reject: (reason: any) => void) => {
       this.setAndEmitConnectionState(ConnectionState.Connecting);
       if (!this.abortController || this.abortController.signal.aborted) {
         this.abortController = new AbortController();
@@ -347,8 +347,11 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
         this.setAndEmitConnectionState(ConnectionState.Connected);
         resolve();
       });
-    }).finally(() => (this.connectFuture = undefined));
-    this.connectFuture = new Future(connectPromise);
+    };
+    this.connectFuture = new Future(connectFn, () => {
+      this.connectFuture = undefined;
+      this.emit(RoomEvent.Connected);
+    });
 
     return this.connectFuture.promise;
   };
@@ -363,7 +366,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       log.warn('abort connection attempt');
       this.abortController?.abort();
       // 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?.reject?.(new ConnectionError('Client initiated disconnect'));
       this.connectFuture = undefined;
     }
     // send leave
@@ -625,7 +628,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     });
     this.setAndEmitConnectionState(ConnectionState.Connected);
     this.emit(RoomEvent.Reconnected);
-    this.reconnectFuture?.resolve();
+    this.reconnectFuture?.resolve?.();
     this.reconnectFuture = undefined;
 
     // rehydrate participants
@@ -672,7 +675,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     }
     // reject potentially ongoing reconnection attempt
     if (this.connectFuture === this.reconnectFuture) {
-      this.connectFuture?.reject(undefined);
+      this.connectFuture?.reject?.(undefined);
       this.connectFuture = undefined;
     }
     this.reconnectFuture = undefined;
@@ -1152,6 +1155,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 export default Room;
 
 export type RoomEventCallbacks = {
+  connected: () => void;
   reconnecting: () => void;
   reconnected: () => void;
   disconnected: (reason?: DisconnectReason) => void;

diff --git a/src/room/events.ts b/src/room/events.ts
@@ -9,6 +9,11 @@
  */
 
 export enum RoomEvent {
+  /**
+   * When the connection to the server has been established
+   */
+  Connected = 'connected',
+
   /**
    * When the connection to the server has been interrupted and it's attempting
    * to reconnect.

diff --git a/src/room/utils.ts b/src/room/utils.ts
@@ -182,16 +182,23 @@ export function getEmptyAudioStreamTrack() {
 export class Future<T> {
   promise: Promise<T>;
 
-  resolve!: (arg: T) => void;
-
-  reject!: (e: any) => void;
-
-  constructor(promise?: Promise<T>) {
-    this.promise =
-      promise ??
-      new Promise<T>((resolve, reject) => {
-        this.resolve = resolve;
-        this.reject = reject;
-      });
+  resolve?: (arg: T) => void;
+
+  reject?: (e: any) => void;
+
+  onFinally?: () => void;
+
+  constructor(
+    futureBase?: (resolve: (arg: T) => void, reject: (e: any) => void) => void,
+    onFinally?: () => void,
+  ) {
+    this.onFinally = onFinally;
+    this.promise = new Promise<T>(async (resolve, reject) => {
+      this.resolve = resolve;
+      this.reject = reject;
+      if (futureBase) {
+        await futureBase(resolve, reject);
+      }
+    }).finally(() => this.onFinally?.());
   }
 }