Switch to 🛎️

src/button/__snapshots__/ReactionToggleButton.test.tsx.snap

@@ -9,7 +9,7 @@ exports[`Can close reaction dialog 1`] = `
     aria-disabled="false"
     aria-expanded="true"
     aria-haspopup="true"
-    aria-labelledby="_r_bb_"
+    aria-labelledby="_r_bh_"
     class="_button_1nw83_8 _has-icon_1nw83_60 _icon-only_1nw83_53"
     data-kind="primary"
     data-size="lg"

src/reactions/index.ts

@@ -29,6 +29,8 @@ import waveSoundOgg from "../sound/reactions/wave.ogg?url";
 import waveSoundMp3 from "../sound/reactions/wave.mp3?url";
 import baduntssSoundOgg from "../sound/reactions/baduntss.ogg?url";
 import baduntssSoundMp3 from "../sound/reactions/baduntss.mp3?url";
+import bellSoundOgg from "../sound/reactions/bell.ogg?url";
+import bellSoundMp3 from "../sound/reactions/bell.mp3?url";
 
 export const ElementCallReactionEventType = "io.element.call.reaction";
 
@@ -202,6 +204,15 @@ export const ReactionSet: ReactionOption[] = [
       mp3: baduntssSoundMp3,
     },
   },
+  {
+    emoji: "🛎️",
+    name: "bell",
+    alias: ["ding"],
+    sound: {
+      ogg: bellSoundOgg,
+      mp3: bellSoundMp3,
+    },
+  },
 ];
 
 export interface RaisedHandInfo {

src/sound/LICENCE.md

@@ -10,6 +10,7 @@ The following sound effects have been licensed from Pixabay, under https://pixab
 - `reactions/generic`
 - `reactions/lightbulb`
 - `reactions/party`
+- `reactions/bell`
 
 ### Other
 

src/state/CallViewModel/CallViewModel.ts

@@ -16,6 +16,7 @@ import {
 import { type Room as MatrixRoom } from "matrix-js-sdk";
 import {
   BehaviorSubject,
+  catchError,
   combineLatest,
   distinctUntilChanged,
   filter,
@@ -38,7 +39,6 @@ import {
   throttleTime,
   timer,
   takeUntil,
-  from,
 } from "rxjs";
 import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
 import {
@@ -112,7 +112,7 @@ import {
   TransportState,
 } from "./localMember/LocalMember.ts";
 import {
-  getLocalTransport,
+  createLocalTransport$,
   type LocalTransport,
 } from "./localMember/LocalTransport.ts";
 import {
@@ -573,16 +573,16 @@ export function createCallViewModel$(
         : `${userId}:${deviceId}`,
   };
 
-  const localTransport$ = options.localTransport
-    ? constant(options.localTransport)
-    : from(
-        getLocalTransport({
-          ownMembershipIdentity,
-          client,
-          roomId: matrixRoom.roomId,
-          matrixRTCMode,
-        }),
-      );
+  const localTransport =
+    options.localTransport ??
+    createLocalTransport$({
+      scope: scope,
+      memberships$: memberships$,
+      ownMembershipIdentity,
+      client,
+      roomId: matrixRoom.roomId,
+      matrixRTCMode,
+    });
 
   const connectionFactory =
     options.connectionFactory ??
@@ -599,7 +599,17 @@ export function createCallViewModel$(
   const connectionManager = createConnectionManager$({
     scope: scope,
     connectionFactory: connectionFactory,
-    localTransport$,
+    localTransport$: scope.behavior(
+      localTransport.active$.pipe(
+        catchError((e: unknown) => {
+          logger.info(
+            "could not pass local transport to createConnectionManager$. localTransport$ threw an error",
+            e,
+          );
+          return of(null);
+        }),
+      ),
+    ),
     remoteTransports$: membershipsAndTransports.transports$,
     logger: logger,
     ownMembershipIdentity,
@@ -655,7 +665,7 @@ export function createCallViewModel$(
     connectionManager,
     client,
     matrixRTCSession,
-    localTransport$,
+    localTransport,
     roomId: matrixRoom.roomId,
     hideScreensharing,
     hostBridge,

src/state/CallViewModel/CallViewModelTestUtils.ts

@@ -210,8 +210,11 @@ export function withCallViewModel(mode: MatrixRTCMode) {
         connectionState$,
         windowSize$,
         localTransport: {
-          transport: exampleTransport,
-          sfuConfig: exampleSfuConfig,
+          active$: constant({
+            transport: exampleTransport,
+            sfuConfig: exampleSfuConfig,
+          }),
+          advertised$: constant(exampleTransport),
         },
         connectionFactory: {
           createConnection(

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

@@ -21,7 +21,7 @@ import {
   beforeEach,
   afterEach,
 } from "vitest";
-import { BehaviorSubject, map, of, Subject } from "rxjs";
+import { BehaviorSubject, map, of } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
 import { type LocalParticipant, type LocalTrack } from "livekit-client";
 import fetchMock from "fetch-mock";
@@ -50,7 +50,10 @@ import {
   TrackState,
   watchScreenShareToggle,
 } from "./LocalMember";
-import { MatrixRTCTransportMissingError } from "../../../utils/errors";
+import {
+  FailToGetOpenIdToken,
+  MatrixRTCTransportMissingError,
+} from "../../../utils/errors";
 import { Epoch, ObservableScope } from "../../ObservableScope";
 import { constant } from "../../Behavior";
 import { ConnectionManagerData } from "../remoteMembers/ConnectionManager";
@@ -58,7 +61,10 @@ import { ConnectionState, type Connection } from "../remoteMembers/Connection";
 import { type Publisher } from "./Publisher";
 import { initializeWidget } from "../../../widget";
 import { nullHostBridge } from "../../../HostBridge";
-import { type LocalTransport } from "./LocalTransport";
+import {
+  type LocalTransport,
+  type LocalTransportWithSFUConfig,
+} from "./LocalTransport";
 import * as openIDSFU from "../../../livekit/openIDSFU";
 
 initializeWidget();
@@ -255,7 +261,7 @@ describe("LocalMembership", () => {
   });
 
   it("throws error on missing RTC config error", () => {
-    withTestScheduler(({ scope, hot, expectObservable }) => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
       const localTransport$ = scope.behavior<null | LivekitTransportConfig>(
         hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")),
         null,
@@ -271,15 +277,16 @@ describe("LocalMembership", () => {
         ),
       };
 
+      const aLocalTransport: LocalTransport = {
+        advertised$: localTransport$,
+        active$: behavior("a", { a: null }),
+      };
+
       const localMembership = createLocalMembership$({
         scope,
         ...defaultCreateLocalMemberValues,
         connectionManager: mockConnectionManager,
-        localTransport$: hot(
-          "1ms #",
-          {},
-          new MatrixRTCTransportMissingError("domain.com"),
-        ),
+        localTransport: aLocalTransport,
       });
 
       expectObservable(localMembership.localMemberState$).toBe("ne", {
@@ -289,9 +296,61 @@ describe("LocalMembership", () => {
     });
   });
 
+  it("Should not publish to active transport if advertised has errors", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      const advertised$ = scope.behavior<null | LivekitTransportConfig>(
+        hot("--#", {}, new FailToGetOpenIdToken(new Error("foo"))),
+        null,
+      );
+
+      // Populate a connection for active
+      const connectionManagerData = new ConnectionManagerData();
+      connectionManagerData.add(connectionTransportBConnected, []);
+      const mockConnectionManager = {
+        transports$: constant(new Epoch([bTransport])),
+        connectionManagerData$: constant(new Epoch(connectionManagerData)),
+      };
+
+      const aLocalTransport: LocalTransport = {
+        advertised$,
+        active$: behavior("a", { n: null, a: bTransportWithSFUConfig }),
+      };
+
+      defaultCreateLocalMemberValues.createPublisherFactory.mockImplementation(
+        () => {
+          return {} as unknown as Publisher;
+        },
+      );
+      const publisherFactory =
+        defaultCreateLocalMemberValues.createPublisherFactory as ReturnType<
+          typeof vi.fn
+        >;
+
+      const localMembership = createLocalMembership$({
+        scope,
+        ...defaultCreateLocalMemberValues,
+        connectionManager: mockConnectionManager,
+        localTransport: aLocalTransport,
+      });
+
+      expectObservable(localMembership.localMemberState$).toBe("n-e", {
+        n: TransportState.Waiting,
+        e: expect.toSatisfy((e) => e instanceof FailToGetOpenIdToken),
+      });
+
+      // Should not have created any publisher
+      expect(publisherFactory).toHaveBeenCalledTimes(0);
+    });
+  });
+
   it("logs if callIntent cannot be updated", async () => {
     const scope = new ObservableScope();
 
+    const aLocalTransport: LocalTransport = {
+      advertised$: constant(aTransport),
+      active$: constant(aTransportWithSFUConfig),
+    };
+
     const mockConnectionManager = {
       transports$: constant(new Epoch([])),
       connectionManagerData$: constant(new Epoch(new ConnectionManagerData())),
@@ -307,7 +366,7 @@ describe("LocalMembership", () => {
         leaveRoomSession: vi.fn(),
       },
       connectionManager: mockConnectionManager,
-      localTransport$: constant(mockTransport),
+      localTransport: aLocalTransport,
     });
     const expextedLog =
       "'not connected yet' while updating the call intent (this is expected on startup)";
@@ -319,19 +378,33 @@ describe("LocalMembership", () => {
     scope.end();
   });
 
-  const mockTransportConfig = {
+  const aTransport = {
     livekit_service_url: "a",
   } as LivekitTransportConfig;
 
-  const mockTransport = {
-    transport: mockTransportConfig,
+  const aTransportWithSFUConfig = {
+    transport: aTransport,
     sfuConfig: {
       jwt: "foo",
       livekitAlias: "bar",
       livekitIdentity: "baz",
       url: "bro",
     },
-  } as LocalTransport;
+  } as LocalTransportWithSFUConfig;
+
+  const bTransport = {
+    livekit_service_url: "b",
+  } as LivekitTransportConfig;
+
+  const bTransportWithSFUConfig = {
+    transport: bTransport,
+    sfuConfig: {
+      jwt: "foo2",
+      livekitAlias: "bar2",
+      livekitIdentity: "baz2",
+      url: "bro2",
+    },
+  } as LocalTransportWithSFUConfig;
 
   const connectionTransportAConnected = {
     livekitRoom: mockLivekitRoom({
@@ -341,13 +414,18 @@ describe("LocalMembership", () => {
       } as unknown as LocalParticipant,
     }),
     state$: constant(ConnectionState.LivekitConnected),
-    transport: mockTransportConfig,
-  } as Connection;
+    transport: aTransport,
+  } as unknown as Connection;
   const connectionTransportAConnecting = {
     ...connectionTransportAConnected,
     state$: constant(ConnectionState.LivekitConnecting),
     livekitRoom: mockLivekitRoom({}),
   } as unknown as Connection;
+  const connectionTransportBConnected = {
+    state$: constant(ConnectionState.LivekitConnected),
+    transport: bTransport,
+    livekitRoom: mockLivekitRoom({}),
+  } as unknown as Connection;
 
   const authCallSpy = vi
     .spyOn(openIDSFU, "getSFUConfigWithOpenID")
@@ -381,7 +459,10 @@ describe("LocalMembership", () => {
           ),
         },
         joinMatrixRTC,
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
         delayId$,
       });
 
@@ -390,7 +471,7 @@ describe("LocalMembership", () => {
       await flushPromises();
       // Joins with timings appropriate for the level of delegation support
       expect(joinMatrixRTC).toHaveBeenCalledWith(
-        mockTransportConfig,
+        aTransport,
         delayedLeaveTimings,
       );
 
@@ -424,7 +505,75 @@ describe("LocalMembership", () => {
     },
   );
 
-  it("only starts tracks if requested", async () => {
+  it("recreates publisher if new connection is used, always unpublish and end tracks", async () => {
+    const scope = new ObservableScope();
+
+    const activeTransport$ = new BehaviorSubject(aTransportWithSFUConfig);
+    const aLocalTransport: LocalTransport = {
+      advertised$: constant(aTransport),
+      active$: activeTransport$,
+    };
+
+    const publishers: Publisher[] = [];
+    let seed = 0;
+    defaultCreateLocalMemberValues.createPublisherFactory.mockImplementation(
+      () => {
+        const a = seed;
+        seed += 1;
+        logger.info(`creating [${a}]`);
+        const p = {
+          // It is enought to check if destroy is called. Destroy itself is tested in the publisher to make sure it does
+          // all the cleanup we need.
+          destroy: vi.fn(),
+          stopPublishing: vi.fn().mockImplementation(() => {
+            logger.info(`stopPublishing [${a}]`);
+          }),
+          stopTracks: vi.fn(),
+        };
+        publishers.push(p as unknown as Publisher);
+        return p;
+      },
+    );
+    const publisherFactory =
+      defaultCreateLocalMemberValues.createPublisherFactory as ReturnType<
+        typeof vi.fn
+      >;
+
+    const connectionManagerData = new ConnectionManagerData();
+    connectionManagerData.add(connectionTransportAConnected, []);
+    connectionManagerData.add(connectionTransportBConnected, []);
+    createLocalMembership$({
+      scope,
+      ...defaultCreateLocalMemberValues,
+      connectionManager: {
+        connectionManagerData$: constant(new Epoch(connectionManagerData)),
+      },
+      localTransport: aLocalTransport,
+    });
+    await flushPromises();
+    activeTransport$.next({
+      ...aTransportWithSFUConfig,
+      transport: bTransport,
+    });
+    await flushPromises();
+
+    expect(publisherFactory).toHaveBeenCalledTimes(2);
+    expect(publishers.length).toBe(2);
+    // stop the first Publisher and let the second one life.
+    expect(publishers[0].destroy).toHaveBeenCalled();
+    expect(publishers[1].destroy).not.toHaveBeenCalled();
+    expect(publisherFactory.mock.calls[0][0].transport).toBe(aTransport);
+    expect(publisherFactory.mock.calls[1][0].transport).toBe(bTransport);
+    scope.end();
+    await flushPromises();
+    // stop all tracks after ending scopes
+    expect(publishers[1].destroy).toHaveBeenCalled();
+    // expect(publishers[1].stopTracks).toHaveBeenCalled();
+
+    defaultCreateLocalMemberValues.createPublisherFactory.mockReset();
+  });
+
+  it("only start tracks if requested", async () => {
     const scope = new ObservableScope();
 
     const publishers: Publisher[] = [];
@@ -453,6 +602,11 @@ describe("LocalMembership", () => {
         typeof vi.fn
       >;
 
+    const aLocalTransport: LocalTransport = {
+      advertised$: constant(aTransport),
+      active$: constant(aTransportWithSFUConfig),
+    };
+
     const connectionManagerData = new ConnectionManagerData();
     connectionManagerData.add(connectionTransportAConnected, []);
     // connectionManagerData.add(connectionTransportB, []);
@@ -462,7 +616,7 @@ describe("LocalMembership", () => {
       connectionManager: {
         connectionManagerData$: constant(new Epoch(connectionManagerData)),
       },
-      localTransport$: constant(mockTransport),
+      localTransport: aLocalTransport,
     });
     await flushPromises();
     expect(publisherFactory).toHaveBeenCalledOnce();
@@ -483,8 +637,16 @@ describe("LocalMembership", () => {
   //
   it("tracks livekit state correctly", async () => {
     const scope = new ObservableScope();
+
     const connectionManagerData = new ConnectionManagerData();
-    const localTransport$ = new Subject<LocalTransport>();
+
+    const activeTransport$ =
+      new BehaviorSubject<null | LocalTransportWithSFUConfig>(null);
+
+    const aLocalTransport: LocalTransport = {
+      advertised$: constant(aTransport),
+      active$: activeTransport$,
+    };
 
     const connectionManagerData$ = new BehaviorSubject(
       new Epoch(connectionManagerData),
@@ -525,14 +687,14 @@ describe("LocalMembership", () => {
       connectionManager: {
         connectionManagerData$,
       },
-      localTransport$,
+      localTransport: aLocalTransport,
     });
 
     await flushPromises();
     expect(localMembership.localMemberState$.value).toStrictEqual(
       TransportState.Waiting,
     );
-    localTransport$.next(mockTransport);
+    activeTransport$.next(aTransportWithSFUConfig);
     await flushPromises();
     expect(localMembership.localMemberState$.value).toStrictEqual({
       matrix: RTCMemberStatus.Connected,
@@ -557,7 +719,7 @@ describe("LocalMembership", () => {
     });
 
     (
-      connectionManagerData2.getConnectionForTransport(mockTransportConfig)!
+      connectionManagerData2.getConnectionForTransport(aTransport)!
         .state$ as BehaviorSubject<ConnectionState>
     ).next(ConnectionState.LivekitConnected);
     expect(localMembership.localMemberState$.value).toStrictEqual({
@@ -660,7 +822,10 @@ describe("LocalMembership", () => {
         connectionManager: {
           connectionManagerData$: constant(new Epoch(connectionManagerData)),
         },
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
       });
 
       await flushPromises();
@@ -697,7 +862,10 @@ describe("LocalMembership", () => {
         connectionManager: {
           connectionManagerData$: constant(new Epoch(connectionManagerData)),
         },
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
       });
 
       await flushPromises();
@@ -745,7 +913,10 @@ describe("LocalMembership", () => {
         connectionManager: {
           connectionManagerData$: constant(new Epoch(connectionManagerData)),
         },
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
       });
 
       await flushPromises();
@@ -786,7 +957,10 @@ describe("LocalMembership", () => {
         connectionManager: {
           connectionManagerData$: constant(new Epoch(connectionManagerData)),
         },
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
       });
 
       await flushPromises();
@@ -852,7 +1026,10 @@ describe("LocalMembership", () => {
         connectionManager: {
           connectionManagerData$: constant(new Epoch(connectionManagerData)),
         },
-        localTransport$: constant(mockTransport),
+        localTransport: {
+          advertised$: constant(aTransport),
+          active$: constant(aTransportWithSFUConfig),
+        },
       });
       return { scope, localMembership };
     };
@@ -862,7 +1039,7 @@ describe("LocalMembership", () => {
       const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
       const connection = {
         state$: constant(ConnectionState.LivekitConnected),
-        transport: mockTransportConfig,
+        transport: aTransport,
         livekitRoom: mockLivekitRoom({
           localParticipant: mockLocalParticipant({
             isScreenShareEnabled: false,

src/state/CallViewModel/localMember/LocalMember.ts

@@ -38,10 +38,6 @@ import {
   startWith,
   switchMap,
   tap,
-  NEVER,
-  concat,
-  race,
-  Subject,
 } from "rxjs";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 import { deepCompare } from "matrix-js-sdk/lib/utils";
@@ -81,6 +77,8 @@ import {
 } from "../remoteMembers/Connection.ts";
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
 import { type LocalTransport } from "./LocalTransport.ts";
+import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
+import { or$ } from "../../../utils/observable.ts";
 import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
 
 export enum TransportState {
@@ -150,7 +148,7 @@ interface Props {
   homeserverConnected: HomeserverConnected;
   roomId: string;
   ownMembershipIdentity: CallMembershipIdentityParts;
-  localTransport$: Observable<LocalTransport>;
+  localTransport: LocalTransport;
   client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
   matrixRTCSession: Pick<
     MatrixRTCSession,
@@ -196,7 +194,7 @@ interface Props {
 export const createLocalMembership$ = ({
   scope,
   connectionManager,
-  localTransport$: localTransportWithErrors$,
+  localTransport,
   homeserverConnected,
   createPublisherFactory,
   joinMatrixRTC,
@@ -254,24 +252,20 @@ export const createLocalMembership$ = ({
   const logger = parentLogger.getChild("[LocalMembership]");
   logger.debug(`Creating local membership..`);
 
-  // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
-  const fatalTransportError$ = new Subject<ElementCallError>();
-  const localTransport$ = localTransportWithErrors$.pipe(
-    catchError((e: unknown) => {
-      let error: ElementCallError;
-      if (e instanceof ElementCallError) {
-        error = e;
-      } else {
-        error = new UnknownCallError(
-          e instanceof Error
-            ? e
-            : new Error("Unknown error from localTransport"),
-        );
-      }
-      fatalTransportError$.next(error);
-      return NEVER; // Make this Observable swallow the error
-    }),
-  );
+  // We consider error on the transport as fatal.
+  // Whether it is the active transport or the preferred transport.
+  const handleTransportError = (e: unknown): Observable<null> => {
+    let error: ElementCallError;
+    if (e instanceof ElementCallError) {
+      error = e;
+    } else {
+      error = new UnknownCallError(
+        e instanceof Error ? e : new Error("Unknown error from localTransport"),
+      );
+    }
+    setTransportError(error);
+    return of(null);
+  };
 
   async function checkDelegationSupport(
     endpointUrl: string,
@@ -309,40 +303,58 @@ export const createLocalMembership$ = ({
 
   // The transport that we will advertise in our membership, paired with info as
   // to whether delayed event delegation is supported
-  const joinParams$ = scope.behavior(
-    localTransport$.pipe(
-      switchMap(async ({ transport }) => {
-        const transportSupportsDelegation = checkDelegationSupport(
-          transport.livekit_service_url + "/delegate_delayed_leave",
-          `transport ${transport.livekit_service_url}`,
-        );
-        return {
-          transport,
-          delegationSupported:
-            (await homeserverSupportsDelegation) ||
-            (await transportSupportsDelegation),
-        };
+  const joinParams$ = localTransport.advertised$.pipe(
+    catchError(handleTransportError),
+    distinctUntilChanged(areLivekitTransportsEqual),
+    switchMap((transport) => {
+      if (transport === null) return of(null);
+      const transportSupportsDelegation = checkDelegationSupport(
+        transport.livekit_service_url + "/delegate_delayed_leave",
+        `transport ${transport.livekit_service_url}`,
+      );
+      return or$(
+        from(homeserverSupportsDelegation),
+        from(transportSupportsDelegation),
+      ).pipe(
+        map((delegationSupported) => ({ transport, delegationSupported })),
+        startWith(null),
+      );
+    }),
+  );
+
+  // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
+  const activeTransport$ = scope.behavior(
+    combineLatest([localTransport.active$, localTransport.advertised$]).pipe(
+      map(([active, advertised]) => {
+        // Our policy is to not publish to another transport if our prefered transport is miss-configured
+        if (advertised == null) return null;
+
+        return active?.transport ?? null;
       }),
+      catchError(handleTransportError),
+      distinctUntilChanged(areLivekitTransportsEqual),
     ),
-    null,
   );
 
   // Drop Epoch data here since we will not combine this anymore
   const localConnection$ = scope.behavior(
     combineLatest([
       connectionManager.connectionManagerData$,
-      localTransport$,
+      activeTransport$,
     ]).pipe(
-      map(([{ value: connectionData }, { transport }]) =>
-        connectionData.getConnectionForTransport(transport),
-      ),
+      map(([{ value: connectionData }, localTransport]) => {
+        if (localTransport === null) {
+          return null;
+        }
+
+        return connectionData.getConnectionForTransport(localTransport);
+      }),
       tap((connection) => {
         logger.info(
           `Local connection updated: ${connection?.transport?.livekit_service_url}`,
         );
       }),
     ),
-    null,
   );
 
   // Tracks error that happen when creating the local tracks.
@@ -478,6 +490,18 @@ export const createLocalMembership$ = ({
     }
   };
 
+  const fatalTransportError$ = new BehaviorSubject<ElementCallError | null>(
+    null,
+  );
+
+  const setTransportError = (e: ElementCallError): void => {
+    if (fatalTransportError$.value !== null) {
+      logger.error("Multiple Transport Errors:", e);
+    } else {
+      fatalTransportError$.next(e);
+    }
+  };
+
   const localConnectionState$ = localConnection$.pipe(
     switchMap((connection) => (connection ? connection.state$ : of(null))),
   );
@@ -485,29 +509,38 @@ export const createLocalMembership$ = ({
   const mediaState$: Behavior<LocalMemberMediaState> = scope.behavior(
     combineLatest([
       localConnectionState$,
+      activeTransport$,
       joinAndPublishRequested$,
       from(trackStartRequested.promise).pipe(
         map(() => true),
         startWith(false),
       ),
     ]).pipe(
-      map(([localConnectionState, shouldPublish, shouldStartTracks]) => {
-        const trackState: TrackState = shouldStartTracks
-          ? TrackState.Ready
-          : TrackState.WaitingForUser;
+      map(
+        ([
+          localConnectionState,
+          localTransport,
+          shouldPublish,
+          shouldStartTracks,
+        ]) => {
+          if (!localTransport) return null;
+          const trackState: TrackState = shouldStartTracks
+            ? TrackState.Ready
+            : TrackState.WaitingForUser;
 
-        if (
-          localConnectionState !== ConnectionState.LivekitConnected ||
-          trackState !== TrackState.Ready
-        )
-          return {
-            connection: localConnectionState,
-            tracks: trackState,
-          };
-        if (!shouldPublish) return PublishState.WaitingForUser;
-        // if (!publishing) return PublishState.Starting;
-        return PublishState.Publishing;
-      }),
+          if (
+            localConnectionState !== ConnectionState.LivekitConnected ||
+            trackState !== TrackState.Ready
+          )
+            return {
+              connection: localConnectionState,
+              tracks: trackState,
+            };
+          if (!shouldPublish) return PublishState.WaitingForUser;
+          // if (!publishing) return PublishState.Starting;
+          return PublishState.Publishing;
+        },
+      ),
       distinctUntilChanged(deepCompare),
     ),
   );
@@ -521,36 +554,30 @@ export const createLocalMembership$ = ({
   };
 
   const localMemberState$ = scope.behavior<LocalMemberState>(
-    concat(
-      // Waiting until…
-      of(TransportState.Waiting),
-      race(
-        // either there is a fatal transport error
-        fatalTransportError$,
-        // or the transport is available.
-        localTransport$.pipe(
-          switchMap(() =>
-            // Once available, track session/media state.
-            combineLatest([
-              mediaState$,
-              homeserverConnected.rtsSession$,
-              fatalMatrixError$,
-              publishError$,
-            ]).pipe(
-              map(
-                ([
-                  mediaState,
-                  rtcSessionStatus,
-                  fatalMatrixError,
-                  publishError,
-                ]) => ({
-                  matrix: fatalMatrixError ?? rtcSessionStatus,
-                  media: publishError ?? mediaState,
-                }),
-              ),
-            ),
-          ),
-        ),
+    combineLatest([
+      mediaState$,
+      homeserverConnected.rtsSession$,
+      fatalMatrixError$,
+      fatalTransportError$,
+      publishError$,
+    ]).pipe(
+      map(
+        ([
+          mediaState,
+          rtcSessionStatus,
+          fatalMatrixError,
+          fatalTransportError,
+          publishError,
+        ]) => {
+          if (fatalTransportError !== null) return fatalTransportError;
+          // `mediaState` will be 'null' until the transport/connection appears.
+          if (mediaState && rtcSessionStatus)
+            return {
+              matrix: fatalMatrixError ?? rtcSessionStatus,
+              media: publishError ?? mediaState,
+            };
+          return TransportState.Waiting;
+        },
       ),
     ),
   );
@@ -658,7 +685,7 @@ export const createLocalMembership$ = ({
       });
   });
 
-  // Join and leave the session as needed
+  // Keep matrix rtc session in sync with advertisedTransport$, connectRequested$
   scope.reconcile(
     scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])),
     async ([joinParams, shouldConnect]) => {

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

@@ -14,10 +14,19 @@ import {
   type MockedObject,
   vi,
 } from "vitest";
+import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
+import { lastValueFrom } from "rxjs";
 import fetchMock from "fetch-mock";
 
-import { mockConfig, ownMemberMock } from "../../../utils/test";
-import { getLocalTransport } from "./LocalTransport";
+import {
+  mockConfig,
+  flushPromises,
+  ownMemberMock,
+  testScope,
+} from "../../../utils/test";
+import { createLocalTransport$ } from "./LocalTransport";
+import { constant } from "../../Behavior";
+import { Epoch, ObservableScope } from "../../ObservableScope";
 import {
   MatrixRTCTransportMissingError,
   FailToGetOpenIdToken,
@@ -38,79 +47,118 @@ describe("LocalTransport", () => {
   beforeEach(() => vi.clearAllMocks());
 
   it("throws if config is missing", async () => {
-    await expect(
-      getLocalTransport({
-        roomId: "!room:example.org",
-        client: {
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getDomain: () => "example.org",
-          // These won't be called in this error path but satisfy the type
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
-      }),
-    ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
+    const { advertised$, active$ } = createLocalTransport$({
+      scope: testScope(),
+      roomId: "!room:example.org",
+      memberships$: constant(new Epoch<CallMembership[]>([])),
+      client: {
+        // eslint-disable-next-line @typescript-eslint/naming-convention
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
+        getDomain: () => "example.org",
+        // These won't be called in this error path but satisfy the type
+        getOpenIdToken: vi.fn(),
+        getDeviceId: vi.fn(),
+      },
+      ownMembershipIdentity: ownMemberMock,
+      matrixRTCMode: MatrixRTCMode.Compatibility,
+    });
+    await flushPromises();
+
+    expect(() => advertised$.value).toThrow(
+      new MatrixRTCTransportMissingError("example.org"),
+    );
+    expect(() => active$.value).toThrow(
+      new MatrixRTCTransportMissingError("example.org"),
+    );
   });
 
   it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
     // Provide a valid config so makeTransportInternal resolves a transport
+    const scope = new ObservableScope();
     mockConfig({
       livekit: { livekit_service_url: "https://lk.example.org" },
     });
-    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
-      new FailToGetOpenIdToken(new Error("no openid")),
+    const resolver = Promise.withResolvers<void>();
+    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockImplementation(
+      async () => {
+        await resolver.promise;
+        throw new FailToGetOpenIdToken(new Error("no openid"));
+      },
     );
+    const observations: unknown[] = [];
+    const errors: Error[] = [];
+    const { advertised$, active$ } = createLocalTransport$({
+      scope,
+      roomId: "!example_room_id",
+      memberships$: constant(new Epoch<CallMembership[]>([])),
+      client: {
+        getDomain: () => "example.org",
+        // eslint-disable-next-line @typescript-eslint/naming-convention
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
+        getOpenIdToken: vi.fn(),
+        getDeviceId: vi.fn(),
+      },
+      ownMembershipIdentity: ownMemberMock,
+      matrixRTCMode: MatrixRTCMode.Compatibility,
+    });
+    active$.subscribe(
+      (o) => observations.push(o),
+      (e) => errors.push(e),
+    );
+    resolver.resolve();
+    await flushPromises();
 
-    await expect(
-      getLocalTransport({
-        roomId: "!example_room_id",
-        client: {
-          getDomain: () => "example.org",
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
-      }),
-    ).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid")));
+    const expectedError = new FailToGetOpenIdToken(new Error("no openid"));
+    expect(observations).toStrictEqual([null]);
+    expect(errors).toStrictEqual([expectedError]);
+    expect(() => advertised$.value).toThrow(expectedError);
+    expect(() => active$.value).toThrow(expectedError);
   });
 
-  it("returns preferred transport", async () => {
+  it("emits preferred transport after OpenID resolves", async () => {
     // Use config so transport discovery succeeds, but delay OpenID JWT fetch
     mockConfig({
       livekit: { livekit_service_url: "https://lk.example.org" },
     });
 
-    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({
+    const openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
+
+    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
+      openIdResolver.promise,
+    );
+
+    const { advertised$, active$ } = createLocalTransport$({
+      scope: testScope(),
+      roomId: "!room:example.org",
+      memberships$: constant(new Epoch<CallMembership[]>([])),
+      client: {
+        // eslint-disable-next-line @typescript-eslint/naming-convention
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
+        getDomain: () => "example.org",
+        getOpenIdToken: vi.fn(),
+        getDeviceId: vi.fn(),
+      },
+      ownMembershipIdentity: ownMemberMock,
+      matrixRTCMode: MatrixRTCMode.Compatibility,
+    });
+
+    openIdResolver.resolve?.({
       url: "https://lk.example.org",
       jwt: "jwt",
       livekitAlias: "Akph4alDMhen",
       livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId,
     });
-
-    expect(
-      await getLocalTransport({
-        roomId: "!room:example.org",
-        client: {
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getDomain: () => "example.org",
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
-      }),
-    ).toStrictEqual({
-      transport: {
-        livekit_service_url: "https://lk.example.org",
-        type: "livekit",
-      },
+    expect(advertised$.value).toBe(null);
+    expect(active$.value).toBe(null);
+    await flushPromises();
+    // final
+    const expectedTransport = {
+      livekit_service_url: "https://lk.example.org",
+      type: "livekit",
+    };
+    expect(advertised$.value).toStrictEqual(expectedTransport);
+    expect(active$.value).toStrictEqual({
+      transport: expectedTransport,
       sfuConfig: {
         jwt: "jwt",
         livekitAlias: "Akph4alDMhen",
@@ -120,19 +168,22 @@ describe("LocalTransport", () => {
     });
   });
 
-  type LocalTransportProps = Parameters<typeof getLocalTransport>[0];
+  type LocalTransportProps = Parameters<typeof createLocalTransport$>[0];
 
   describe("transport configuration mechanisms", () => {
     let localTransportOpts: LocalTransportProps & {
       client: MockedObject<LocalTransportProps["client"]>;
     };
+    let openIdResolver: PromiseWithResolvers<openIDSFU.SFUConfig>;
     beforeEach(() => {
       mockConfig({});
       customLivekitUrl.setValue(customLivekitUrl.defaultValue);
       localTransportOpts = {
         ownMembershipIdentity: ownMemberMock,
+        scope: testScope(),
         roomId: "!example_room_id",
         matrixRTCMode: MatrixRTCMode.Compatibility,
+        memberships$: constant(new Epoch<CallMembership[]>([])),
         client: {
           getDomain: vi.fn().mockReturnValue("example.org"),
           // eslint-disable-next-line @typescript-eslint/naming-convention
@@ -141,6 +192,10 @@ describe("LocalTransport", () => {
           getDeviceId: vi.fn(),
         },
       };
+      openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
+      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
+        openIdResolver.promise,
+      );
     });
 
     afterEach(() => {
@@ -151,15 +206,19 @@ describe("LocalTransport", () => {
       mockConfig({
         livekit: { livekit_service_url: "https://lk.example.org" },
       });
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
-
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
-        transport: {
-          livekit_service_url: "https://lk.example.org",
-          type: "livekit",
-        },
+      const { advertised$, active$ } =
+        createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(advertised$.value).toBe(null);
+      expect(active$.value).toBe(null);
+      await flushPromises();
+      const expectedTransport = {
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      };
+      expect(advertised$.value).toStrictEqual(expectedTransport);
+      expect(active$.value).toStrictEqual({
+        transport: expectedTransport,
         sfuConfig: {
           jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
           livekitAlias: "Akph4alDMhen",
@@ -171,11 +230,12 @@ describe("LocalTransport", () => {
 
     it("supports getting transport via user settings", async () => {
       customLivekitUrl.setValue("https://lk.example.org");
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
-
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      const { advertised$, active$ } =
+        createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(advertised$.value).toBe(null);
+      await flushPromises();
+      expect(active$.value).toStrictEqual({
         transport: {
           livekit_service_url: "https://lk.example.org",
           type: "livekit",
@@ -193,15 +253,19 @@ describe("LocalTransport", () => {
       localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
         { type: "livekit", livekit_service_url: "https://lk.example.org" },
       ]);
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
-
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
-        transport: {
-          livekit_service_url: "https://lk.example.org",
-          type: "livekit",
-        },
+      const { advertised$, active$ } =
+        createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(advertised$.value).toBe(null);
+      expect(active$.value).toBe(null);
+      await flushPromises();
+      const expectedTransport = {
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      };
+      expect(advertised$.value).toStrictEqual(expectedTransport);
+      expect(active$.value).toStrictEqual({
+        transport: expectedTransport,
         sfuConfig: {
           jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
           livekitAlias: "Akph4alDMhen",
@@ -215,31 +279,38 @@ describe("LocalTransport", () => {
       localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
         { type: "livekit", livekit_service_url: "https://lk.example.org" },
       ]);
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
+      openIdResolver.reject(
         new FailToGetOpenIdToken(new Error("Test driven error")),
       );
-
-      await expect(getLocalTransport(localTransportOpts)).rejects.toThrow(
-        expect.any(FailToGetOpenIdToken),
-      );
+      await expect(async () =>
+        lastValueFrom(createLocalTransport$(localTransportOpts).active$),
+      ).rejects.toThrow(expect.any(FailToGetOpenIdToken));
     });
 
     it("throws if no options are available", async () => {
-      await expect(
-        getLocalTransport({
-          ownMembershipIdentity: ownMemberMock,
-          roomId: "!example_room_id",
-          matrixRTCMode: MatrixRTCMode.Compatibility,
-          client: {
-            getDomain: () => "example.org",
-            // eslint-disable-next-line @typescript-eslint/naming-convention
-            _unstable_getRTCTransports: async () => Promise.resolve([]),
-            // These won't be called in this error path but satisfy the type
-            getOpenIdToken: vi.fn(),
-            getDeviceId: vi.fn(),
-          },
-        }),
-      ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
+      const { advertised$, active$ } = createLocalTransport$({
+        scope: testScope(),
+        ownMembershipIdentity: ownMemberMock,
+        roomId: "!example_room_id",
+        matrixRTCMode: MatrixRTCMode.Compatibility,
+        memberships$: constant(new Epoch<CallMembership[]>([])),
+        client: {
+          getDomain: () => "example.org",
+          // eslint-disable-next-line @typescript-eslint/naming-convention
+          _unstable_getRTCTransports: async () => Promise.resolve([]),
+          // These won't be called in this error path but satisfy the type
+          getOpenIdToken: vi.fn(),
+          getDeviceId: vi.fn(),
+        },
+      });
+      await flushPromises();
+
+      expect(() => advertised$.value).toThrow(
+        new MatrixRTCTransportMissingError("example.org"),
+      );
+      expect(() => active$.value).toThrow(
+        new MatrixRTCTransportMissingError("example.org"),
+      );
     });
   });
 });

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -5,11 +5,17 @@ SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
+import {
+  type CallMembership,
+  type LivekitTransportConfig,
+} from "matrix-js-sdk/lib/matrixrtc";
 import { type MatrixClient } from "matrix-js-sdk";
+import { distinctUntilChanged, from, map, of, switchMap } from "rxjs";
 import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
 import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
 
+import { type Behavior } from "../../Behavior.ts";
+import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
 import { Config } from "../../../config/Config.ts";
 import {
   FailToGetOpenIdToken,
@@ -21,12 +27,19 @@ import {
   type SFUConfig,
   type OpenIDClientParts,
 } from "../../../livekit/openIDSFU.ts";
+import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
 import { customLivekitUrl } from "../../../settings/settings.ts";
 import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
 import { type MatrixRTCMode } from "../../../config/ConfigOptions.ts";
 
+/*
+ * It figures out “which LiveKit focus URL/alias the local user should use,”
+ * and ensures the SFU path is primed before advertising that choice.
+ */
 interface Props {
+  scope: ObservableScope;
   ownMembershipIdentity: CallMembershipIdentityParts;
+  memberships$: Behavior<Epoch<CallMembership[]>>;
   client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
     OpenIDClientParts;
   // Used by the jwt service to create the livekit room and compute the livekit alias.
@@ -55,65 +68,114 @@ interface Props {
 //
 // 2.
 // We need to make sure we do not sent livekit_alias in sticky events and that we drop all code for sending state events!
-export interface LocalTransport {
+export interface LocalTransportWithSFUConfig {
   transport: LivekitTransportConfig;
   sfuConfig: SFUConfig;
 }
 
-export function isLocalTransport(
-  obj: LivekitTransportConfig | LocalTransport,
-): obj is LocalTransport {
+export function isLocalTransportWithSFUConfig(
+  obj: LivekitTransportConfig | LocalTransportWithSFUConfig,
+): obj is LocalTransportWithSFUConfig {
   return "transport" in obj && "sfuConfig" in obj;
 }
 
+export interface LocalTransport {
+  /**
+   * The transport to be advertised in our MatrixRTC membership. `null` when not
+   * yet fetched/validated.
+   */
+  advertised$: Behavior<LivekitTransportConfig | null>;
+  /**
+   * The transport to connect to and publish media on. `null` when not yet known
+   * or available.
+   */
+  active$: Behavior<LocalTransportWithSFUConfig | null>;
+}
+
 /**
- * Connects to the JWT service and determines the transport that the local member should use.
+ * Connects to the JWT service and determines the transports that the local member should use.
  *
  * @prop useOldJwtEndpoint Whether to set forceOldJwtEndpoint on the returned transport and to use the old JWT endpoint.
  * This is used when the connection manager needs to know if it has to use the legacy endpoint which implies a string concatenated rtcBackendIdentity.
  * (which is expected for non sticky event based rtc member events)
- * @returns The transport to advertise in our MatrixRTC membership and publish media on.
+ * @returns The transport to advertise in the local MatrixRTC membership, along with the transport to actively publish media to.
  * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
  */
-export async function getLocalTransport({
+export const createLocalTransport$ = ({
+  scope,
+  memberships$,
   ownMembershipIdentity,
   client,
   roomId,
   matrixRTCMode,
-}: Props): Promise<LocalTransport> {
+}: Props): LocalTransport => {
   const logger = rootLogger.getChild("[LocalTransport]");
-  const discovery = new RtcTransportAutoDiscovery({
+
+  const transportDiscovery = new RtcTransportAutoDiscovery({
     client: client,
     resolvedConfig: Config.get(),
     logger: logger,
   });
-  const customUrl = customLivekitUrl.value$.value;
 
-  // Respect the user's custom URL, if set
-  const transport: LivekitTransportConfig | null = customUrl
-    ? { type: "livekit", livekit_service_url: customUrl }
-    : await discovery.discoverPreferredTransport();
+  // Get the preferred transport from the current deployment.
+  const discoveredTransport$ = from(
+    transportDiscovery.discoverPreferredTransport(),
+  );
 
-  if (transport === null)
-    throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
+  const preferredConfig$ = customLivekitUrl.value$.pipe(
+    switchMap((customUrl) => {
+      if (customUrl) {
+        return of({
+          type: "livekit",
+          livekit_service_url: customUrl,
+        } as LivekitTransportConfig);
+      } else {
+        return discoveredTransport$;
+      }
+    }),
+    map((config) => {
+      if (!config) {
+        // Bubbled up from the preferredConfig$ observable.
+        throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
+      }
+      return config;
+    }),
+    distinctUntilChanged(areLivekitTransportsEqual),
+  );
 
-  try {
-    return await doOpenIdAndJWTFromUrl(
-      transport,
-      matrixRTCMode,
-      ownMembershipIdentity,
-      roomId,
-      client,
-      logger,
-    );
-  } catch (e) {
-    logger.error(
-      `Failed to authenticate to transport ${transport.livekit_service_url}`,
-      e,
-    );
-    throw mapAuthErrorToUserFriendlyError(e);
-  }
-}
+  const preferredTransport$ = preferredConfig$.pipe(
+    switchMap(async (transport) => {
+      try {
+        return await doOpenIdAndJWTFromUrl(
+          transport,
+          matrixRTCMode,
+          ownMembershipIdentity,
+          roomId,
+          client,
+          logger,
+        );
+      } catch (e) {
+        logger.error(
+          `Failed to authenticate to transport ${transport.livekit_service_url}`,
+          e,
+        );
+        throw mapAuthErrorToUserFriendlyError(e);
+      }
+    }),
+  );
+
+  // Always publish on and advertise the preferred transport.
+  return {
+    advertised$: scope.behavior(
+      preferredTransport$.pipe(
+        map((t) => t.transport),
+        distinctUntilChanged(areLivekitTransportsEqual),
+      ),
+      null,
+    ),
+    active$: scope.behavior(preferredTransport$, null),
+  };
+};
 
 /**
  *  Utility to ensure the user can authenticate with the SFU.
@@ -137,7 +199,7 @@ async function doOpenIdAndJWTFromUrl(
   roomId: string,
   client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts,
   logger?: Logger,
-): Promise<LocalTransport> {
+): Promise<LocalTransportWithSFUConfig> {
   const sfuConfig = await getSFUConfigWithOpenID(
     client,
     membership,

src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
 import { type RemoteParticipant } from "livekit-client";
 import { logger } from "matrix-js-sdk/lib/logger";
@@ -20,7 +20,7 @@ import { type ConnectionFactory } from "./ConnectionFactory.ts";
 import { type Connection } from "./Connection.ts";
 import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
 import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
-import { type Behavior } from "../../Behavior.ts";
+import { constant, type Behavior } from "../../Behavior.ts";
 
 // Some test constants
 
@@ -74,7 +74,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport$: constant(null),
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
@@ -115,7 +115,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport$: constant(null),
         remoteTransports$: behavior("abcdef", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1], 1),
@@ -162,7 +162,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport$: constant(null),
         remoteTransports$: behavior("abc", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
@@ -297,7 +297,7 @@ describe("connectionManagerData$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport$: constant(null),
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),

src/state/CallViewModel/remoteMembers/ConnectionManager.ts

@@ -7,16 +7,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
-import {
-  combineLatest,
-  map,
-  type Observable,
-  of,
-  switchMap,
-  startWith,
-  catchError,
-  NEVER,
-} from "rxjs";
+import { combineLatest, map, of, switchMap } from "rxjs";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 import { type RemoteParticipant } from "livekit-client";
 import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
@@ -28,8 +19,8 @@ import { generateItemsWithEpoch } from "../../../utils/observable.ts";
 import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
 import {
-  isLocalTransport,
-  type LocalTransport,
+  isLocalTransportWithSFUConfig,
+  type LocalTransportWithSFUConfig,
 } from "../localMember/LocalTransport.ts";
 import { type SFUConfig } from "../../../livekit/openIDSFU.ts";
 
@@ -87,7 +78,7 @@ export class ConnectionManagerData {
 interface Props {
   scope: ObservableScope;
   connectionFactory: ConnectionFactory;
-  localTransport$: Observable<LocalTransport>;
+  localTransport$: Behavior<LocalTransportWithSFUConfig | null>;
   remoteTransports$: Behavior<Epoch<LivekitTransportConfig[]>>;
 
   logger: Logger;
@@ -130,14 +121,6 @@ export function createConnectionManager$({
   const logger = parentLogger.getChild("[ConnectionManager]");
   // TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
 
-  const localTransportAsArray$ = localTransport$.pipe(
-    // LocalMember already surfaces local transport errors properly in the UI,
-    // here we can just swallow them
-    catchError(() => NEVER),
-    map((transport) => [transport]),
-    startWith([]),
-  );
-
   /**
    * All transports currently managed by the ConnectionManager.
    *
@@ -147,12 +130,16 @@ export function createConnectionManager$({
    * externally this is modified via `registerTransports()`.
    */
   const localAndRemoteTransports$: Behavior<
-    Epoch<(LivekitTransportConfig | LocalTransport)[]>
+    Epoch<(LivekitTransportConfig | LocalTransportWithSFUConfig)[]>
   > = scope.behavior(
-    combineLatest([localTransportAsArray$, remoteTransports$]).pipe(
+    combineLatest([remoteTransports$, localTransport$]).pipe(
       // Combine local and remote transports into one transport array
       // and set the forceOldJwtEndpoint property on the local transport
-      map(([localTransportAsArray, remoteTransports]) => {
+      map(([remoteTransports, localTransport]) => {
+        let localTransportAsArray: LocalTransportWithSFUConfig[] = [];
+        if (localTransport) {
+          localTransportAsArray = [localTransport];
+        }
         const dedupedRemote = removeDuplicateTransports(remoteTransports.value);
         const remoteWithoutLocal = dedupedRemote.filter(
           (transport) =>
@@ -183,9 +170,8 @@ export function createConnectionManager$({
         "ConnectionManager connections$",
         function* (transports) {
           for (const transport of transports) {
-            if (isLocalTransport(transport)) {
-              // This is the local transport; only the `LocalTransport`
-              // interface has a `sfuConfig` field.
+            if (isLocalTransportWithSFUConfig(transport)) {
+              // This is the local transport; only the `LocalTransportWithSFUConfig` has a `sfuConfig` field.
               yield {
                 keys: [
                   transport.transport.livekit_service_url,

src/state/CallViewModel/remoteMembers/integration.test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { test, vi, expect, beforeEach, afterEach } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type Room as LivekitRoom } from "livekit-client";
 import EventEmitter from "events";
 import fetchMock from "fetch-mock";
@@ -34,6 +34,7 @@ import {
 } from "./MatrixLivekitMembers.ts";
 import { createConnectionManager$ } from "./ConnectionManager.ts";
 import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
+import { constant } from "../../Behavior.ts";
 import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
 
 // Test the integration of ConnectionManager and MatrixLivekitMerger
@@ -123,7 +124,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
     const connectionManager = createConnectionManager$({
       scope: testScope,
       connectionFactory: ecConnectionFactory,
-      localTransport$: NEVER,
+      localTransport$: constant(null),
       remoteTransports$: membershipsAndTransports.transports$,
       logger: logger,
       ownMembershipIdentity: ownMemberMock,

src/state/SessionBehaviors.ts

@@ -22,10 +22,6 @@ import {
 } from "./ObservableScope";
 import { type Behavior } from "./Behavior";
 
-/**
- * Tracks the transports used by ourselves, plus all other MatrixRTC session
- * members.
- */
 export const membershipsAndTransports$ = (
   scope: ObservableScope,
   memberships$: Behavior<Epoch<CallMembership[]>>,
@@ -35,6 +31,14 @@ export const membershipsAndTransports$ = (
   >;
   transports$: Behavior<Epoch<LivekitTransportConfig[]>>;
 } => {
+  /**
+   * Lists the transports used by ourselves, plus all other MatrixRTC session
+   * members.
+   * For completeness this also lists the preferred transport and
+   * whether we are in multi-SFU mode or sticky events mode.
+   * `advertisedTransport$` reads these values together, so bundling them avoids inconsistent state or
+   * excessive updates when using RxJS.
+   */
   const membershipsWithTransport$: Behavior<
     Epoch<
       {

src/utils/observable.test.ts

@@ -9,9 +9,30 @@ import { expect, test } from "vitest";
 import { type Observable, of, Subject, switchMap } from "rxjs";
 
 import { withTestScheduler } from "./test";
-import { filterBehavior, generateItems, pauseWhen } from "./observable";
+import { or$, filterBehavior, generateItems, pauseWhen } from "./observable";
 import { type Behavior } from "../state/Behavior";
 
+const yesNo = {
+  y: true,
+  n: false,
+};
+
+test("or$", () => {
+  withTestScheduler(({ behavior, expectObservable }) => {
+    const input1Marbles = "ny--n--";
+    const input2Marbles = "n-y--n-";
+    const input3Marbles = "n--y--n";
+    const outputMarbles = "nyyyyyn";
+    expectObservable(
+      or$(
+        behavior(input1Marbles, yesNo),
+        behavior(input2Marbles, yesNo),
+        behavior(input3Marbles, yesNo),
+      ),
+    ).toBe(outputMarbles, yesNo);
+  });
+});
+
 test("pauseWhen", () => {
   withTestScheduler(({ behavior, expectObservable }) => {
     const inputMarbles = " abcdefgh-i-jk-";

src/utils/observable.ts

@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
 import {
   type Observable,
   audit,
+  combineLatest,
   concat,
   defer,
   filter,
@@ -112,6 +113,14 @@ export function getValue<T>(state$: Observable<T>): T {
   return value;
 }
 
+/**
+ * Creates an Observable that has a value of true whenever some of its inputs
+ * are true.
+ */
+export function or$(...inputs: Observable<boolean>[]): Observable<boolean> {
+  return combineLatest(inputs, (...flags) => flags.some((flag) => flag));
+}
+
 /**
  * RxJS operator that pauses all changes in the input value whenever a Behavior
  * is true. When the Behavior returns to being false, the most recently