FCE-205 Refactor into smaller files

e2e-tests/ts-client/app/package.json

@@ -36,6 +36,7 @@
     "eslint-plugin-react-refresh": "^0.4.5",
     "testcontainers": "^10.3.2",
     "ts-proto": "^1.176.0",
+    "typescript": "^5.5.0",
     "vite": "^5.1.2",
     "vitest": "^1.6.0"
   }

e2e-tests/ts-client/app/src/MockComponent.tsx

@@ -86,8 +86,8 @@ export const MockComponent = ({ webrtc }: Props) => {
     const track = stream.getVideoTracks()[0];
 
     const simulcastConfig: SimulcastConfig = {
-      enabled: false,
-      activeEncodings: [],
+      enabled: true,
+      activeEncodings: ["h", "m", "l"],
       disabledEncodings: [],
     };
     const maxBandwidth: BandwidthLimit = 0;

e2e-tests/ts-client/app/src/VideoPlayerWithDetector.tsx

@@ -46,7 +46,7 @@ export const VideoPlayerWithDetector = ({ stream, id, webrtc }: Props) => {
   }, [stream]);
 
   const getDecodedFrames = useCallback(async () => {
-    const connection = webrtc["connection"];
+    const connection = webrtc["stateManager"]["connection"];
     if (!connection) return 0;
 
     const inbound = getTrackIdentifierToInboundRtp(await connection.getStats());

e2e-tests/ts-client/scenarios/utils.ts

@@ -137,14 +137,14 @@ export const assertThatRemoteTracksAreVisible = async (
   });
 };
 
-type WebrtcClient = { webrtc?: { connection?: RTCPeerConnection } };
+type WebrtcClient = { webrtc?: { stateManager: { connection?: RTCPeerConnection } } };
 type WindowType = typeof window;
 
 export const assertThatOtherVideoIsPlaying = async (page: Page) => {
   await test.step('Assert that media is working', async () => {
     const getDecodedFrames: () => Promise<number> = () =>
       page.evaluate(async () => {
-        const connection = (window as WindowType & WebrtcClient)?.webrtc
+        const connection = (window as WindowType & WebrtcClient)?.webrtc?.stateManager
           ?.connection;
         // connection object is available after first renegotiation (sdpOffer, sdpAnswer)
         if (!window || !connection) return -1;

packages/react-client/package.json

@@ -42,7 +42,7 @@
     "typecheck": "tsc"
   },
   "devDependencies": {
-    "@playwright/test": "^1.42.1",
+    "@playwright/test": "^1.45.1",
     "@types/events": "^3.0.3",
     "@types/lodash.isequal": "^4.5.8",
     "@types/node": "^20.14.10",

packages/ts-client/package.json

@@ -53,7 +53,7 @@
     "uuid": "^10.0.0"
   },
   "devDependencies": {
-    "@playwright/test": "^1.40.1",
+    "@playwright/test": "^1.45.1",
     "@types/events": "^3.0.3",
     "@types/node": "^20.14.10",
     "@types/uuid": "^10.0.0",

packages/ts-client/src/webrtc/CommandsQueue.ts

@@ -0,0 +1,156 @@
+import type { Command } from './commands';
+import type { Deferred } from './deferred';
+import type { StateManager } from './StateManager';
+import type { NegotiationManager } from './NegotiationManager';
+
+export class CommandsQueue<EndpointMetadata, TrackMetadata> {
+  private readonly stateManager: StateManager<EndpointMetadata, TrackMetadata>;
+  private readonly negotiationManager: NegotiationManager;
+
+  private clearConnectionCallbacks: (() => void) | null = null;
+
+  constructor(
+    stateManager: StateManager<EndpointMetadata, TrackMetadata>,
+    negotiationManager: NegotiationManager,
+  ) {
+    this.stateManager = stateManager;
+    this.negotiationManager = negotiationManager;
+  }
+
+  public setupEventListeners = (connection: RTCPeerConnection) => {
+    const onSignalingStateChange = () => {
+      switch (this.stateManager.connection?.signalingState) {
+        case 'stable':
+          this.processNextCommand();
+          break;
+      }
+    };
+
+    const onIceGatheringStateChange = () => {
+      switch (this.stateManager.connection?.iceGatheringState) {
+        case 'complete':
+          this.processNextCommand();
+          break;
+      }
+    };
+
+    const onConnectionStateChange = () => {
+      switch (connection.connectionState) {
+        case 'connected':
+          this.processNextCommand();
+          break;
+      }
+    };
+    const onIceConnectionStateChange = () => {
+      switch (this.stateManager.connection?.iceConnectionState) {
+        case 'connected':
+          this.processNextCommand();
+          break;
+      }
+    };
+
+    this.clearConnectionCallbacks = () => {
+      connection.removeEventListener(
+        'signalingstatechange',
+        onSignalingStateChange,
+      );
+      connection.removeEventListener(
+        'icegatheringstatechange',
+        onIceGatheringStateChange,
+      );
+      connection.removeEventListener(
+        'connectionstatechange',
+        onConnectionStateChange,
+      );
+      connection.removeEventListener(
+        'iceconnectionstatechange',
+        onIceConnectionStateChange,
+      );
+    };
+
+    connection.addEventListener(
+      'icegatheringstatechange',
+      onIceConnectionStateChange,
+    );
+    connection.addEventListener(
+      'connectionstatechange',
+      onConnectionStateChange,
+    );
+    connection.addEventListener(
+      'iceconnectionstatechange',
+      onIceConnectionStateChange,
+    );
+    connection.addEventListener('signalingstatechange', onSignalingStateChange);
+  };
+
+  private commandsQueue: Command[] = [];
+  private commandResolutionNotifier: Deferred<void> | null = null;
+
+  public pushCommand = (command: Command) => {
+    this.commandsQueue.push(command);
+    this.processNextCommand();
+  };
+
+  private isConnectionUnstable = () => {
+    const connection = this.stateManager.connection;
+    if (connection === undefined) return false;
+
+    const isSignalingUnstable = connection.signalingState !== 'stable';
+    const isConnectionNotConnected = connection.connectionState !== 'connected';
+    const isIceNotConnected = connection.iceConnectionState !== 'connected';
+
+    return isSignalingUnstable && isConnectionNotConnected && isIceNotConnected;
+  };
+
+  private isNegotiationInProgress = () => {
+    return (
+      this.negotiationManager.ongoingRenegotiation ||
+      this.stateManager.ongoingTrackReplacement
+    );
+  };
+
+  public processNextCommand = () => {
+    if (this.isNegotiationInProgress()) return;
+    if (this.isConnectionUnstable()) return;
+
+    this.resolvePreviousCommand();
+
+    const command = this.commandsQueue.shift();
+
+    if (!command) return;
+
+    this.commandResolutionNotifier = command.resolutionNotifier;
+    this.handleCommand(command);
+  };
+
+  private handleCommand = (command: Command) => {
+    const error = command.validate?.();
+
+    if (error) {
+      this.commandResolutionNotifier?.reject(error);
+      this.commandResolutionNotifier = null;
+      this.processNextCommand();
+    } else {
+      command.handler();
+
+      if (command.resolve === 'immediately') {
+        this.resolvePreviousCommand();
+        this.processNextCommand();
+      }
+    }
+  };
+
+  private resolvePreviousCommand = () => {
+    if (!this.commandResolutionNotifier) return;
+
+    this.commandResolutionNotifier.resolve();
+    this.commandResolutionNotifier = null;
+  };
+
+  public cleanUp = () => {
+    this.commandResolutionNotifier?.reject('Disconnected');
+    this.commandResolutionNotifier = null;
+    this.commandsQueue = [];
+    this.clearConnectionCallbacks?.();
+  };
+}

packages/ts-client/src/webrtc/NegotiationManager.ts

@@ -0,0 +1,7 @@
+export class NegotiationManager {
+  /**
+   * Indicates if an ongoing renegotiation is active.
+   * During renegotiation, both parties are expected to actively exchange events: renegotiateTracks, offerData, sdpOffer, sdpAnswer.
+   */
+  public ongoingRenegotiation: boolean = false;
+}

packages/ts-client/src/webrtc/RTCPeerConnectionUtils.ts

@@ -5,3 +5,14 @@ export const findSender = (
   connection!
     .getSenders()
     .find((sender) => sender.track && sender!.track!.id === trackId)!;
+
+export const findSenderByTrack = (
+  connection: RTCPeerConnection | undefined,
+  track: MediaStreamTrack | null | undefined,
+): RTCRtpSender | undefined =>
+  connection?.getSenders().find((sender) => sender.track === track);
+
+export const isTrackInUse = (
+  connection: RTCPeerConnection | undefined,
+  track: MediaStreamTrack,
+) => connection?.getSenders().some((val) => val.track === track);