diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index b2c763c2e..1ab4ea134 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -16,7 +16,6 @@ import { import { type Room as MatrixRoom } from "matrix-js-sdk"; import { BehaviorSubject, - catchError, combineLatest, distinctUntilChanged, filter, @@ -39,6 +38,7 @@ 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 { - createLocalTransport$, + getLocalTransport, type LocalTransport, } from "./localMember/LocalTransport.ts"; import { @@ -573,16 +573,16 @@ export function createCallViewModel$( : `${userId}:${deviceId}`, }; - const localTransport = - options.localTransport ?? - createLocalTransport$({ - scope: scope, - memberships$: memberships$, - ownMembershipIdentity, - client, - roomId: matrixRoom.roomId, - matrixRTCMode, - }); + const localTransport$ = options.localTransport + ? constant(options.localTransport) + : from( + getLocalTransport({ + ownMembershipIdentity, + client, + roomId: matrixRoom.roomId, + matrixRTCMode, + }), + ); const connectionFactory = options.connectionFactory ?? @@ -599,17 +599,7 @@ export function createCallViewModel$( const connectionManager = createConnectionManager$({ scope: scope, connectionFactory: connectionFactory, - 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); - }), - ), - ), + localTransport$, remoteTransports$: membershipsAndTransports.transports$, logger: logger, ownMembershipIdentity, @@ -665,7 +655,7 @@ export function createCallViewModel$( connectionManager, client, matrixRTCSession, - localTransport, + localTransport$, roomId: matrixRoom.roomId, hideScreensharing, hostBridge, diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index 5a3fa3378..0863ea2b7 100644 --- a/src/state/CallViewModel/CallViewModelTestUtils.ts +++ b/src/state/CallViewModel/CallViewModelTestUtils.ts @@ -210,11 +210,8 @@ export function withCallViewModel(mode: MatrixRTCMode) { connectionState$, windowSize$, localTransport: { - active$: constant({ - transport: exampleTransport, - sfuConfig: exampleSfuConfig, - }), - advertised$: constant(exampleTransport), + transport: exampleTransport, + sfuConfig: exampleSfuConfig, }, connectionFactory: { createConnection( diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 6df999b1e..973c3f6c0 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -21,7 +21,7 @@ import { beforeEach, afterEach, } from "vitest"; -import { BehaviorSubject, map, of } from "rxjs"; +import { BehaviorSubject, map, of, Subject } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; import { type LocalParticipant, type LocalTrack } from "livekit-client"; import fetchMock from "fetch-mock"; @@ -50,10 +50,7 @@ import { TrackState, watchScreenShareToggle, } from "./LocalMember"; -import { - FailToGetOpenIdToken, - MatrixRTCTransportMissingError, -} from "../../../utils/errors"; +import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { Epoch, ObservableScope } from "../../ObservableScope"; import { constant } from "../../Behavior"; import { ConnectionManagerData } from "../remoteMembers/ConnectionManager"; @@ -61,10 +58,7 @@ import { ConnectionState, type Connection } from "../remoteMembers/Connection"; import { type Publisher } from "./Publisher"; import { initializeWidget } from "../../../widget"; import { nullHostBridge } from "../../../HostBridge"; -import { - type LocalTransport, - type LocalTransportWithSFUConfig, -} from "./LocalTransport"; +import { type LocalTransport } from "./LocalTransport"; import * as openIDSFU from "../../../livekit/openIDSFU"; initializeWidget(); @@ -261,7 +255,7 @@ describe("LocalMembership", () => { }); it("throws error on missing RTC config error", () => { - withTestScheduler(({ scope, hot, behavior, expectObservable }) => { + withTestScheduler(({ scope, hot, expectObservable }) => { const localTransport$ = scope.behavior( hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")), null, @@ -277,16 +271,15 @@ describe("LocalMembership", () => { ), }; - const aLocalTransport: LocalTransport = { - advertised$: localTransport$, - active$: behavior("a", { a: null }), - }; - const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport: aLocalTransport, + localTransport$: hot( + "1ms #", + {}, + new MatrixRTCTransportMissingError("domain.com"), + ), }); expectObservable(localMembership.localMemberState$).toBe("ne", { @@ -296,61 +289,9 @@ describe("LocalMembership", () => { }); }); - it("Should not publish to active transport if advertised has errors", () => { - withTestScheduler(({ scope, hot, behavior, expectObservable }) => { - const advertised$ = scope.behavior( - 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())), @@ -366,7 +307,7 @@ describe("LocalMembership", () => { leaveRoomSession: vi.fn(), }, connectionManager: mockConnectionManager, - localTransport: aLocalTransport, + localTransport$: constant(mockTransport), }); const expextedLog = "'not connected yet' while updating the call intent (this is expected on startup)"; @@ -378,33 +319,19 @@ describe("LocalMembership", () => { scope.end(); }); - const aTransport = { + const mockTransportConfig = { livekit_service_url: "a", } as LivekitTransportConfig; - const aTransportWithSFUConfig = { - transport: aTransport, + const mockTransport = { + transport: mockTransportConfig, sfuConfig: { jwt: "foo", livekitAlias: "bar", livekitIdentity: "baz", url: "bro", }, - } 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; + } as LocalTransport; const connectionTransportAConnected = { livekitRoom: mockLivekitRoom({ @@ -414,18 +341,13 @@ describe("LocalMembership", () => { } as unknown as LocalParticipant, }), state$: constant(ConnectionState.LivekitConnected), - transport: aTransport, - } as unknown as Connection; + transport: mockTransportConfig, + } 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") @@ -459,10 +381,7 @@ describe("LocalMembership", () => { ), }, joinMatrixRTC, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), delayId$, }); @@ -471,7 +390,7 @@ describe("LocalMembership", () => { await flushPromises(); // Joins with timings appropriate for the level of delegation support expect(joinMatrixRTC).toHaveBeenCalledWith( - aTransport, + mockTransportConfig, delayedLeaveTimings, ); @@ -505,75 +424,7 @@ describe("LocalMembership", () => { }, ); - 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 () => { + it("only starts tracks if requested", async () => { const scope = new ObservableScope(); const publishers: Publisher[] = []; @@ -602,11 +453,6 @@ describe("LocalMembership", () => { typeof vi.fn >; - const aLocalTransport: LocalTransport = { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }; - const connectionManagerData = new ConnectionManagerData(); connectionManagerData.add(connectionTransportAConnected, []); // connectionManagerData.add(connectionTransportB, []); @@ -616,7 +462,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: aLocalTransport, + localTransport$: constant(mockTransport), }); await flushPromises(); expect(publisherFactory).toHaveBeenCalledOnce(); @@ -637,16 +483,8 @@ describe("LocalMembership", () => { // it("tracks livekit state correctly", async () => { const scope = new ObservableScope(); - const connectionManagerData = new ConnectionManagerData(); - - const activeTransport$ = - new BehaviorSubject(null); - - const aLocalTransport: LocalTransport = { - advertised$: constant(aTransport), - active$: activeTransport$, - }; + const localTransport$ = new Subject(); const connectionManagerData$ = new BehaviorSubject( new Epoch(connectionManagerData), @@ -687,14 +525,14 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$, }, - localTransport: aLocalTransport, + localTransport$, }); await flushPromises(); expect(localMembership.localMemberState$.value).toStrictEqual( TransportState.Waiting, ); - activeTransport$.next(aTransportWithSFUConfig); + localTransport$.next(mockTransport); await flushPromises(); expect(localMembership.localMemberState$.value).toStrictEqual({ matrix: RTCMemberStatus.Connected, @@ -719,7 +557,7 @@ describe("LocalMembership", () => { }); ( - connectionManagerData2.getConnectionForTransport(aTransport)! + connectionManagerData2.getConnectionForTransport(mockTransportConfig)! .state$ as BehaviorSubject ).next(ConnectionState.LivekitConnected); expect(localMembership.localMemberState$.value).toStrictEqual({ @@ -822,10 +660,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), }); await flushPromises(); @@ -862,10 +697,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), }); await flushPromises(); @@ -913,10 +745,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), }); await flushPromises(); @@ -957,10 +786,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), }); await flushPromises(); @@ -1026,10 +852,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport: { - advertised$: constant(aTransport), - active$: constant(aTransportWithSFUConfig), - }, + localTransport$: constant(mockTransport), }); return { scope, localMembership }; }; @@ -1039,7 +862,7 @@ describe("LocalMembership", () => { const setScreenShareEnabled = vi.fn().mockRejectedValue(error); const connection = { state$: constant(ConnectionState.LivekitConnected), - transport: aTransport, + transport: mockTransportConfig, livekitRoom: mockLivekitRoom({ localParticipant: mockLocalParticipant({ isScreenShareEnabled: false, diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 7303946a0..bf5080902 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -38,6 +38,10 @@ 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"; @@ -77,8 +81,6 @@ 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 { @@ -148,7 +150,7 @@ interface Props { homeserverConnected: HomeserverConnected; roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; - localTransport: LocalTransport; + localTransport$: Observable; client: Pick; matrixRTCSession: Pick< MatrixRTCSession, @@ -194,7 +196,7 @@ interface Props { export const createLocalMembership$ = ({ scope, connectionManager, - localTransport, + localTransport$: localTransportWithErrors$, homeserverConnected, createPublisherFactory, joinMatrixRTC, @@ -252,20 +254,24 @@ export const createLocalMembership$ = ({ const logger = parentLogger.getChild("[LocalMembership]"); logger.debug(`Creating local membership..`); - // We consider error on the transport as fatal. - // Whether it is the active transport or the preferred transport. - const handleTransportError = (e: unknown): Observable => { - 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); - }; + // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error. + const fatalTransportError$ = new Subject(); + 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 + }), + ); async function checkDelegationSupport( endpointUrl: string, @@ -303,58 +309,40 @@ 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$ = 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; + 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), + }; }), - catchError(handleTransportError), - distinctUntilChanged(areLivekitTransportsEqual), ), + null, ); // Drop Epoch data here since we will not combine this anymore const localConnection$ = scope.behavior( combineLatest([ connectionManager.connectionManagerData$, - activeTransport$, + localTransport$, ]).pipe( - map(([{ value: connectionData }, localTransport]) => { - if (localTransport === null) { - return null; - } - - return connectionData.getConnectionForTransport(localTransport); - }), + map(([{ value: connectionData }, { transport }]) => + connectionData.getConnectionForTransport(transport), + ), tap((connection) => { logger.info( `Local connection updated: ${connection?.transport?.livekit_service_url}`, ); }), ), + null, ); // Tracks error that happen when creating the local tracks. @@ -490,18 +478,6 @@ export const createLocalMembership$ = ({ } }; - const fatalTransportError$ = new BehaviorSubject( - 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))), ); @@ -509,38 +485,29 @@ export const createLocalMembership$ = ({ const mediaState$: Behavior = scope.behavior( combineLatest([ localConnectionState$, - activeTransport$, joinAndPublishRequested$, from(trackStartRequested.promise).pipe( map(() => true), startWith(false), ), ]).pipe( - map( - ([ - localConnectionState, - localTransport, - shouldPublish, - shouldStartTracks, - ]) => { - if (!localTransport) return null; - const trackState: TrackState = shouldStartTracks - ? TrackState.Ready - : TrackState.WaitingForUser; + map(([localConnectionState, shouldPublish, shouldStartTracks]) => { + 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), ), ); @@ -554,30 +521,36 @@ export const createLocalMembership$ = ({ }; const localMemberState$ = scope.behavior( - 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; - }, + 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, + }), + ), + ), + ), + ), ), ), ); diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index a77c7a16d..90d0d15e5 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -14,19 +14,10 @@ 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, - flushPromises, - ownMemberMock, - testScope, -} from "../../../utils/test"; -import { createLocalTransport$ } from "./LocalTransport"; -import { constant } from "../../Behavior"; -import { Epoch, ObservableScope } from "../../ObservableScope"; +import { mockConfig, ownMemberMock } from "../../../utils/test"; +import { getLocalTransport } from "./LocalTransport"; import { MatrixRTCTransportMissingError, FailToGetOpenIdToken, @@ -47,118 +38,79 @@ describe("LocalTransport", () => { beforeEach(() => vi.clearAllMocks()); it("throws if config is missing", async () => { - const { advertised$, active$ } = createLocalTransport$({ - scope: testScope(), - roomId: "!room:example.org", - memberships$: constant(new Epoch([])), - 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"), - ); + 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")); }); 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" }, }); - const resolver = Promise.withResolvers(); - vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockImplementation( - async () => { - await resolver.promise; - throw new FailToGetOpenIdToken(new Error("no openid")); - }, + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue( + 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([])), - 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(); - 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); + 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"))); }); - it("emits preferred transport after OpenID resolves", async () => { + it("returns preferred transport", async () => { // Use config so transport discovery succeeds, but delay OpenID JWT fetch mockConfig({ livekit: { livekit_service_url: "https://lk.example.org" }, }); - const openIdResolver = Promise.withResolvers(); - - vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue( - openIdResolver.promise, - ); - - const { advertised$, active$ } = createLocalTransport$({ - scope: testScope(), - roomId: "!room:example.org", - memberships$: constant(new Epoch([])), - 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?.({ + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({ url: "https://lk.example.org", jwt: "jwt", livekitAlias: "Akph4alDMhen", livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId, }); - 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, + + 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", + }, sfuConfig: { jwt: "jwt", livekitAlias: "Akph4alDMhen", @@ -168,22 +120,19 @@ describe("LocalTransport", () => { }); }); - type LocalTransportProps = Parameters[0]; + type LocalTransportProps = Parameters[0]; describe("transport configuration mechanisms", () => { let localTransportOpts: LocalTransportProps & { client: MockedObject; }; - let openIdResolver: PromiseWithResolvers; beforeEach(() => { mockConfig({}); customLivekitUrl.setValue(customLivekitUrl.defaultValue); localTransportOpts = { ownMembershipIdentity: ownMemberMock, - scope: testScope(), roomId: "!example_room_id", matrixRTCMode: MatrixRTCMode.Compatibility, - memberships$: constant(new Epoch([])), client: { getDomain: vi.fn().mockReturnValue("example.org"), // eslint-disable-next-line @typescript-eslint/naming-convention @@ -192,10 +141,6 @@ describe("LocalTransport", () => { getDeviceId: vi.fn(), }, }; - openIdResolver = Promise.withResolvers(); - vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue( - openIdResolver.promise, - ); }); afterEach(() => { @@ -206,19 +151,15 @@ describe("LocalTransport", () => { mockConfig({ livekit: { livekit_service_url: "https://lk.example.org" }, }); - 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, + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue( + openIdResponse, + ); + + expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ + transport: { + livekit_service_url: "https://lk.example.org", + type: "livekit", + }, sfuConfig: { jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", livekitAlias: "Akph4alDMhen", @@ -230,12 +171,11 @@ describe("LocalTransport", () => { it("supports getting transport via user settings", async () => { customLivekitUrl.setValue("https://lk.example.org"); - const { advertised$, active$ } = - createLocalTransport$(localTransportOpts); - openIdResolver.resolve?.(openIdResponse); - expect(advertised$.value).toBe(null); - await flushPromises(); - expect(active$.value).toStrictEqual({ + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue( + openIdResponse, + ); + + expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ transport: { livekit_service_url: "https://lk.example.org", type: "livekit", @@ -253,19 +193,15 @@ describe("LocalTransport", () => { localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ { type: "livekit", livekit_service_url: "https://lk.example.org" }, ]); - 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, + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue( + openIdResponse, + ); + + expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ + transport: { + livekit_service_url: "https://lk.example.org", + type: "livekit", + }, sfuConfig: { jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", livekitAlias: "Akph4alDMhen", @@ -279,38 +215,31 @@ describe("LocalTransport", () => { localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ { type: "livekit", livekit_service_url: "https://lk.example.org" }, ]); - openIdResolver.reject( + vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue( new FailToGetOpenIdToken(new Error("Test driven error")), ); - await expect(async () => - lastValueFrom(createLocalTransport$(localTransportOpts).active$), - ).rejects.toThrow(expect.any(FailToGetOpenIdToken)); + + await expect(getLocalTransport(localTransportOpts)).rejects.toThrow( + expect.any(FailToGetOpenIdToken), + ); }); it("throws if no options are available", async () => { - const { advertised$, active$ } = createLocalTransport$({ - scope: testScope(), - ownMembershipIdentity: ownMemberMock, - roomId: "!example_room_id", - matrixRTCMode: MatrixRTCMode.Compatibility, - memberships$: constant(new Epoch([])), - 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"), - ); + 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")); }); }); }); diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index c5aa39590..0aa150d5d 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -5,17 +5,11 @@ SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -import { - type CallMembership, - type LivekitTransportConfig, -} from "matrix-js-sdk/lib/matrixrtc"; +import { 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, @@ -27,19 +21,12 @@ 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>; client: Pick & OpenIDClientParts; // Used by the jwt service to create the livekit room and compute the livekit alias. @@ -68,114 +55,65 @@ 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 LocalTransportWithSFUConfig { +export interface LocalTransport { transport: LivekitTransportConfig; sfuConfig: SFUConfig; } -export function isLocalTransportWithSFUConfig( - obj: LivekitTransportConfig | LocalTransportWithSFUConfig, -): obj is LocalTransportWithSFUConfig { +export function isLocalTransport( + obj: LivekitTransportConfig | LocalTransport, +): obj is LocalTransport { 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; - /** - * The transport to connect to and publish media on. `null` when not yet known - * or available. - */ - active$: Behavior; -} - /** - * Connects to the JWT service and determines the transports that the local member should use. + * Connects to the JWT service and determines the transport 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 the local MatrixRTC membership, along with the transport to actively publish media to. + * @returns The transport to advertise in our MatrixRTC membership and publish media on. * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken */ -export const createLocalTransport$ = ({ - scope, - memberships$, +export async function getLocalTransport({ ownMembershipIdentity, client, roomId, matrixRTCMode, -}: Props): LocalTransport => { +}: Props): Promise { const logger = rootLogger.getChild("[LocalTransport]"); - - const transportDiscovery = new RtcTransportAutoDiscovery({ + const discovery = new RtcTransportAutoDiscovery({ client: client, resolvedConfig: Config.get(), logger: logger, }); + const customUrl = customLivekitUrl.value$.value; - // Get the preferred transport from the current deployment. - const discoveredTransport$ = from( - transportDiscovery.discoverPreferredTransport(), - ); + // Respect the user's custom URL, if set + const transport: LivekitTransportConfig | null = customUrl + ? { type: "livekit", livekit_service_url: customUrl } + : await discovery.discoverPreferredTransport(); - 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), - ); + if (transport === null) + throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); - 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), - }; -}; + 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); + } +} /** * Utility to ensure the user can authenticate with the SFU. @@ -199,7 +137,7 @@ async function doOpenIdAndJWTFromUrl( roomId: string, client: Pick & OpenIDClientParts, logger?: Logger, -): Promise { +): Promise { const sfuConfig = await getSFUConfigWithOpenID( client, membership, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts index 809c8bccb..0903702ff 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts +++ b/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 } from "rxjs"; +import { BehaviorSubject, NEVER } 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 { constant, type Behavior } from "../../Behavior.ts"; +import { type Behavior } from "../../Behavior.ts"; // Some test constants @@ -74,7 +74,7 @@ describe("connections$ stream", () => { const { connectionManagerData$ } = createConnectionManager$({ scope: testScope, connectionFactory: fakeConnectionFactory, - localTransport$: constant(null), + localTransport$: NEVER, 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$: constant(null), + localTransport$: NEVER, 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$: constant(null), + localTransport$: NEVER, 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$: constant(null), + localTransport$: NEVER, remoteTransports$: behavior("a", { a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0), }), diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts index 5ebd2ce6d..4af08b49b 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts @@ -7,7 +7,14 @@ Please see LICENSE in the repository root for full details. */ import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc"; -import { combineLatest, map, of, switchMap } from "rxjs"; +import { + combineLatest, + map, + type Observable, + of, + switchMap, + startWith, +} 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"; @@ -19,8 +26,8 @@ import { generateItemsWithEpoch } from "../../../utils/observable.ts"; import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { - isLocalTransportWithSFUConfig, - type LocalTransportWithSFUConfig, + isLocalTransport, + type LocalTransport, } from "../localMember/LocalTransport.ts"; import { type SFUConfig } from "../../../livekit/openIDSFU.ts"; @@ -78,7 +85,7 @@ export class ConnectionManagerData { interface Props { scope: ObservableScope; connectionFactory: ConnectionFactory; - localTransport$: Behavior; + localTransport$: Observable; remoteTransports$: Behavior>; logger: Logger; @@ -121,6 +128,11 @@ 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( + map((transport) => [transport]), + startWith([]), + ); + /** * All transports currently managed by the ConnectionManager. * @@ -130,16 +142,12 @@ export function createConnectionManager$({ * externally this is modified via `registerTransports()`. */ const localAndRemoteTransports$: Behavior< - Epoch<(LivekitTransportConfig | LocalTransportWithSFUConfig)[]> + Epoch<(LivekitTransportConfig | LocalTransport)[]> > = scope.behavior( - combineLatest([remoteTransports$, localTransport$]).pipe( + combineLatest([localTransportAsArray$, remoteTransports$]).pipe( // Combine local and remote transports into one transport array // and set the forceOldJwtEndpoint property on the local transport - map(([remoteTransports, localTransport]) => { - let localTransportAsArray: LocalTransportWithSFUConfig[] = []; - if (localTransport) { - localTransportAsArray = [localTransport]; - } + map(([localTransportAsArray, remoteTransports]) => { const dedupedRemote = removeDuplicateTransports(remoteTransports.value); const remoteWithoutLocal = dedupedRemote.filter( (transport) => @@ -170,7 +178,7 @@ export function createConnectionManager$({ "ConnectionManager connections$", function* (transports) { for (const transport of transports) { - if (isLocalTransportWithSFUConfig(transport)) { + if (isLocalTransport(transport)) { // This is the local transport; only the `LocalTransportWithSFUConfig` has a `sfuConfig` field. yield { keys: [ diff --git a/src/state/CallViewModel/remoteMembers/integration.test.ts b/src/state/CallViewModel/remoteMembers/integration.test.ts index 67d15a38f..0db25c3b5 100644 --- a/src/state/CallViewModel/remoteMembers/integration.test.ts +++ b/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 } from "rxjs"; +import { BehaviorSubject, NEVER } from "rxjs"; import { type Room as LivekitRoom } from "livekit-client"; import EventEmitter from "events"; import fetchMock from "fetch-mock"; @@ -34,7 +34,6 @@ 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 @@ -124,7 +123,7 @@ test("bob, carl, then bob joining no tracks yet", () => { const connectionManager = createConnectionManager$({ scope: testScope, connectionFactory: ecConnectionFactory, - localTransport$: constant(null), + localTransport$: NEVER, remoteTransports$: membershipsAndTransports.transports$, logger: logger, ownMembershipIdentity: ownMemberMock, diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index 8f09cb6ef..d73adfda7 100644 --- a/src/state/ObservableScope.ts +++ b/src/state/ObservableScope.ts @@ -18,6 +18,7 @@ import { share, take, takeUntil, + shareReplay, } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; @@ -67,6 +68,17 @@ export class ObservableScope { public readonly share: MonoTypeOperator = (input$) => input$.pipe(this.bindImpl, this.shareImpl); + private readonly shareReplayImpl: MonoTypeOperator = shareReplay({ + bufferSize: 1, + refCount: false, + }); + /** + * Shares (multicasts) the Observable as a hot Observable, replaying the most + * recently emitted value upon subscription. + */ + public readonly shareReplay: MonoTypeOperator = (input$) => + input$.pipe(this.bindImpl, this.shareReplayImpl); + /** * Converts an Observable to a Behavior. If no initial value is specified, the * Observable must synchronously emit an initial value. diff --git a/src/utils/observable.test.ts b/src/utils/observable.test.ts index 392f742ab..80cbb3c83 100644 --- a/src/utils/observable.test.ts +++ b/src/utils/observable.test.ts @@ -9,30 +9,9 @@ import { expect, test } from "vitest"; import { type Observable, of, Subject, switchMap } from "rxjs"; import { withTestScheduler } from "./test"; -import { or$, filterBehavior, generateItems, pauseWhen } from "./observable"; +import { 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-"; diff --git a/src/utils/observable.ts b/src/utils/observable.ts index 8468b3fbb..ca51ad196 100644 --- a/src/utils/observable.ts +++ b/src/utils/observable.ts @@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details. import { type Observable, audit, - combineLatest, concat, defer, filter, @@ -113,14 +112,6 @@ export function getValue(state$: Observable): T { return value; } -/** - * Creates an Observable that has a value of true whenever some of its inputs - * are true. - */ -export function or$(...inputs: Observable[]): Observable { - 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