Another larger refactor to fix sfu switches and in general proper

cleanup.

Author: toger5

locales/en/app.json

@@ -108,11 +108,14 @@
     "connection_lost_description": "You were disconnected from the call.",
     "e2ee_unsupported": "Incompatible browser",
     "e2ee_unsupported_description": "Your web browser does not support encrypted calls. Supported browsers include Chrome, Safari, and Firefox 117+.",
+    "failed_to_start_livekit": "Failed to start Livekit",
     "generic": "Something went wrong",
     "generic_description": "Submitting debug logs will help us track down the problem.",
     "insufficient_capacity": "Insufficient capacity",
     "insufficient_capacity_description": "The server has reached its maximum capacity and you cannot join the call at this time. Try again later, or contact your server admin if the problem persists.",
     "matrix_rtc_transport_missing": "The server is not configured to work with {{brand}}. Please contact your server admin (Domain: {{domain}}, Error Code: {{ errorCode }}).",
+    "membership_manager": "Membership Manager Error",
+    "membership_manager_description": "The Membership Manager had to shut down. This is caused by many consequtive failed network requests.",
     "open_elsewhere": "Opened in another tab",
     "open_elsewhere_description": "{{brand}} has been opened in another tab. If that doesn't sound right, try reloading the page.",
     "room_creation_restricted": "Failed to create call",

src/room/InCallView.tsx

@@ -146,6 +146,8 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
       reactionsReader.reactions$,
       scope.behavior(trackProcessorState$),
     );
+    // TODO move this somewhere else once we use the callViewModel in the lobby as well!
+    vm.join();
     setVm(vm);
 
     vm.leave$.pipe(scope.bind()).subscribe(props.onLeft);

src/state/CallViewModel/CallViewModel.ts

@@ -102,7 +102,6 @@ import {
   createLocalMembership$,
   enterRTCSession,
   LivekitState,
-  type LocalMemberConnectionState,
 } from "./localMember/LocalMembership.ts";
 import { createLocalTransport$ } from "./localMember/LocalTransport.ts";
 import {
@@ -202,7 +201,7 @@ export interface CallViewModel {
   hangup: () => void;
 
   // joining
-  join: () => LocalMemberConnectionState;
+  join: () => void;
 
   // screen sharing
   /**
@@ -572,15 +571,6 @@ export function createCallViewModel$(
     ),
   );
 
-  // CODESMELL?
-  // This is functionally the same Observable as leave$, except here it's
-  // hoisted to the top of the class. This enables the cyclic dependency between
-  // leave$ -> autoLeave$ -> callPickupState$ -> livekitConnectionState$ ->
-  // localConnection$ -> transports$ -> joined$ -> leave$.
-  const leaveHoisted$ = new Subject<
-    "user" | "timeout" | "decline" | "allOthersLeft"
-  >();
-
   /**
    * Whether various media/event sources should pretend to be disconnected from
    * all network input, even if their connection still technically works.
@@ -840,10 +830,7 @@ export function createCallViewModel$(
     merge(
       autoLeave$,
       merge(userHangup$, widgetHangup$).pipe(map(() => "user" as const)),
-    ).pipe(
-      scope.share,
-      tap((reason) => leaveHoisted$.next(reason)),
-    );
+    ).pipe(scope.share);
 
   const spotlightSpeaker$ = scope.behavior<UserMediaViewModel | null>(
     userMedia$.pipe(
@@ -1448,16 +1435,13 @@ export function createCallViewModel$(
   // reassigned here to make it publicly accessible
   const toggleScreenSharing = localMembership.toggleScreenSharing;
 
-  const join = localMembership.requestConnect;
-  // TODO-MULTI-SFU: Use this view model for the lobby as well, and only call this once 'join' is clicked?
-  join();
   return {
     autoLeave$: autoLeave$,
     callPickupState$: callPickupState$,
     ringOverlay$: ringOverlay$,
     leave$: leave$,
     hangup: (): void => userHangup$.next(),
-    join: join,
+    join: localMembership.requestConnect,
     toggleScreenSharing: toggleScreenSharing,
     sharingScreen$: sharingScreen$,
 

src/state/CallViewModel/localMember/LocalMembership.test.ts

@@ -12,12 +12,15 @@ import {
 } from "matrix-js-sdk/lib/matrixrtc";
 import { describe, expect, it, vi } from "vitest";
 import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
-import { map } from "rxjs";
+import { BehaviorSubject, map, of } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
+import { type LocalParticipant } from "livekit-client";
 
 import { MatrixRTCMode } from "../../../settings/settings";
 import {
+  flushPromises,
   mockConfig,
+  mockLivekitRoom,
   mockMuteStates,
   withTestScheduler,
 } from "../../../utils/test";
@@ -27,14 +30,19 @@ import {
   LivekitState,
 } from "./LocalMembership";
 import { MatrixRTCTransportMissingError } from "../../../utils/errors";
-import { Epoch } from "../../ObservableScope";
+import { Epoch, ObservableScope } from "../../ObservableScope";
 import { constant } from "../../Behavior";
 import { ConnectionManagerData } from "../remoteMembers/ConnectionManager";
-import { type Publisher } from "./Publisher";
+import { type Connection } from "../remoteMembers/Connection";
 
 const MATRIX_RTC_MODE = MatrixRTCMode.Legacy;
 const getUrlParams = vi.hoisted(() => vi.fn(() => ({})));
 vi.mock("../../../UrlParams", () => ({ getUrlParams }));
+vi.mock("@livekit/components-core", () => ({
+  observeParticipantEvents: vi
+    .fn()
+    .mockReturnValue(of({ isScreenShareEnabled: false })),
+}));
 
 describe("LocalMembership", () => {
   describe("enterRTCSession", () => {
@@ -183,7 +191,7 @@ describe("LocalMembership", () => {
       processor: undefined,
     }),
     logger: logger,
-    createPublisherFactory: (): Publisher => ({}) as unknown as Publisher,
+    createPublisherFactory: vi.fn(),
     joinMatrixRTC: async (): Promise<void> => {},
     homeserverConnected$: constant(true),
   };
@@ -216,7 +224,7 @@ describe("LocalMembership", () => {
       });
 
       expectObservable(localMembership.connectionState.livekit$).toBe("ne", {
-        n: { state: LivekitState.Uninitialized },
+        n: { state: LivekitState.Connecting },
         e: {
           state: LivekitState.Error,
           error: expect.toSatisfy(
@@ -226,4 +234,63 @@ describe("LocalMembership", () => {
       });
     });
   });
+
+  it("recreates publisher if new connection is used", async () => {
+    const scope = new ObservableScope();
+    const aTransport = {
+      livekit_service_url: "a",
+    } as LivekitTransport;
+    const bTransport = {
+      livekit_service_url: "b",
+    } as LivekitTransport;
+
+    const localTransport$ = new BehaviorSubject(aTransport);
+
+    const connectionManagerData = new ConnectionManagerData();
+
+    connectionManagerData.add(
+      {
+        livekitRoom: mockLivekitRoom({
+          localParticipant: {
+            isScreenShareEnabled: false,
+            trackPublications: [],
+          } as unknown as LocalParticipant,
+        }),
+        state$: constant({
+          state: "ConnectedToLkRoom",
+        }),
+        transport: aTransport,
+      } as unknown as Connection,
+      [],
+    );
+    connectionManagerData.add(
+      {
+        state$: constant({
+          state: "ConnectedToLkRoom",
+        }),
+        transport: bTransport,
+      } as unknown as Connection,
+      [],
+    );
+
+    const publisherFactory =
+      defaultCreateLocalMemberValues.createPublisherFactory as ReturnType<
+        typeof vi.fn
+      >;
+
+    createLocalMembership$({
+      scope,
+      ...defaultCreateLocalMemberValues,
+      connectionManager: {
+        connectionManagerData$: constant(new Epoch(connectionManagerData)),
+      },
+      localTransport$,
+    });
+    await flushPromises();
+    localTransport$.next(bTransport);
+    await flushPromises();
+    expect(publisherFactory).toHaveBeenCalledTimes(2);
+    expect(publisherFactory.mock.calls[0][0].transport).toBe(aTransport);
+    expect(publisherFactory.mock.calls[1][0].transport).toBe(bTransport);
+  });
 });