diff --git a/src/room/InCallView.tsx b/src/room/InCallView.tsx index 51c916f92..8b66eb160 100644 --- a/src/room/InCallView.tsx +++ b/src/room/InCallView.tsx @@ -694,15 +694,13 @@ export const InCallView: FC = ({ tab={settingsTab} onTabChange={setSettingsTab} developerSettingsVm={developerSettingsVm} - livekitRooms={allConnections - .getConnections() - .map((connectionItem) => ({ - room: connectionItem.livekitRoom, - livekitAlias: connectionItem.livekitAlias, - // TODO compute is local or tag it in the livekit room items already - isLocal: undefined, - url: connectionItem.transport.livekit_service_url, - }))} + livekitRooms={allConnections.map((connectionItem) => ({ + room: connectionItem.livekitRoom, + livekitAlias: connectionItem.livekitAlias, + // TODO compute is local or tag it in the livekit room items already + isLocal: undefined, + url: connectionItem.transport.livekit_service_url, + }))} /> )} diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 216a1e617..a005c05ed 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -39,6 +39,9 @@ import { timer, takeUntil, from, + concatWith, + catchError, + switchAll, } from "rxjs"; import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger"; import { @@ -56,6 +59,7 @@ import { createToggle$, filterBehavior, generateItems, + mapScoped, pauseWhen, } from "../../utils/observable"; import { @@ -105,7 +109,7 @@ import { type OneOnOneMedia, } from "../layoutMedia.ts"; import { ElementCallError, UnknownCallError } from "../../utils/errors.ts"; -import { type Epoch, type ObservableScope } from "../ObservableScope.ts"; +import { Epoch, type ObservableScope } from "../ObservableScope.ts"; import { createHomeserverConnected$ } from "./localMember/HomeserverConnected.ts"; import { createLocalMembership$, @@ -123,9 +127,9 @@ import { ECConnectionFactory, } from "./remoteMembers/ConnectionFactory.ts"; import { - type ConnectionManagerData, - createConnectionManager$, -} from "./remoteMembers/ConnectionManager.ts"; + ConnectionMap, + createConnectionMap$, +} from "./remoteMembers/ConnectionMap.ts"; import { createRemoteMatrixLivekitMembers$, type LocalMatrixLivekitMember, @@ -371,7 +375,7 @@ export interface CallViewModel { * key is generated when someone joins or leaves. */ keyRotationSuppressed$: Behavior; - allConnections$: Behavior; + allConnections$: Behavior; /** Participants sorted by livekit room so they can be used in the audio rendering */ livekitRoomItems$: Behavior; /** use the layout instead, this is just for the sdk export. */ @@ -577,14 +581,24 @@ export function createCallViewModel$( options.livekitRoomFactory, ); - const connectionManager = createConnectionManager$({ - scope: scope, - connectionFactory: connectionFactory, - localTransport$, - remoteTransports$: membershipsAndTransports.transports$, - logger: logger, - ownMembershipIdentity, - }); + const connectionMap$ = scope.behavior>( + localTransport$.pipe( + catchError(() => NEVER), // Swallow errors + concatWith(NEVER), // So the Observable doesn't complete prematurely + mapScoped("connectionMap$", (localTransport, scope) => + createConnectionMap$({ + scope, + connectionFactory, + localTransport, + remoteTransports$: membershipsAndTransports.transports$, + logger, + ownMembershipIdentity, + }), + ), + switchAll(), + ), + new Epoch(new ConnectionMap()), + ); const remoteMatrixLivekitMembers$: Behavior< Epoch @@ -592,7 +606,7 @@ export function createCallViewModel$( scope: scope, membershipsWithTransport$: membershipsAndTransports.membershipsWithTransport$, - connectionManager: connectionManager, + connectionMap$, localUser: { userId, deviceId }, }); @@ -631,7 +645,7 @@ export function createCallViewModel$( controlledAudioDevices, ); }, - connectionManager, + connectionMap$, client, matrixRTCSession, localTransport$, @@ -717,7 +731,7 @@ export function createCallViewModel$( }); const allConnections$ = scope.behavior( - connectionManager.connectionManagerData$.pipe(map((d) => d.value)), + connectionMap$.pipe(map((d) => d.value.getConnections())), ); const livekitRoomItems$ = scope.behavior( diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index b1e309d58..543cada8c 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, Subject } from "rxjs"; +import { BehaviorSubject, of, Subject } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; import { ConnectionState as LivekitConnectionState, @@ -57,7 +57,7 @@ import { import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { Epoch, ObservableScope } from "../../ObservableScope"; import { constant } from "../../Behavior"; -import { ConnectionManagerData } from "../remoteMembers/ConnectionManager"; +import { ConnectionMap } from "../remoteMembers/ConnectionMap"; import { type ConnectionState, type Connection, @@ -261,25 +261,11 @@ describe("LocalMembership", () => { it("throws error on missing RTC config error", () => { withTestScheduler(({ scope, hot, expectObservable }) => { - const localTransport$ = scope.behavior( - hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")), - null, - ); - - // we do not need any connection data since we want to fail before reaching that. - const mockConnectionManager = { - transports$: scope.behavior( - localTransport$.pipe(map((t) => new Epoch([t]))), - ), - connectionManagerData$: constant( - new Epoch(new ConnectionManagerData()), - ), - }; - const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, - connectionManager: mockConnectionManager, + // we do not need any connection data since we want to fail before reaching that. + connectionMap$: constant(new Epoch(new ConnectionMap())), localTransport$: hot( "1ms #", {}, @@ -295,24 +281,18 @@ describe("LocalMembership", () => { }); it("logs if callIntent cannot be updated", async () => { - const scope = new ObservableScope(); - - const mockConnectionManager = { - transports$: constant(new Epoch([])), - connectionManagerData$: constant(new Epoch(new ConnectionManagerData())), - }; async function reject(): Promise { return Promise.reject(new Error("Not connected yet")); } const localMembership = createLocalMembership$({ - scope, + scope: testScope(), ...defaultCreateLocalMemberValues, matrixRTCSession: { slotId: "m.call#ROOM", updateCallIntent: vi.fn().mockImplementation(reject), leave: vi.fn(), }, - connectionManager: mockConnectionManager, + connectionMap$: constant(new Epoch(new ConnectionMap())), localTransport$: constant(mockTransport), }); const expextedLog = @@ -322,7 +302,6 @@ describe("LocalMembership", () => { await flushPromises(); defaultCreateLocalMemberValues.muteStates.video.setEnabled$.value?.(true); expect(internalLogger).toHaveBeenCalledWith(expextedLog); - scope.end(); }); const mockTransport = { @@ -374,14 +353,12 @@ describe("LocalMembership", () => { if (delegationUrl !== null) fetchMock.post(delegationUrl, () => ({ status: 401, body: {} })); - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(connectionTransportAConnected, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(connectionTransportAConnected, []); const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), joinMatrixRTC, localTransport$: constant(mockTransport), delayId$, @@ -453,15 +430,13 @@ describe("LocalMembership", () => { typeof vi.fn >; - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(connectionTransportAConnected, []); - // connectionManagerData.add(connectionTransportB, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(connectionTransportAConnected, []); + // connectionMap.add(connectionTransportB, []); const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); await flushPromises(); @@ -483,12 +458,10 @@ describe("LocalMembership", () => { // it("tracks livekit state correctly", async () => { const scope = new ObservableScope(); - const connectionManagerData = new ConnectionManagerData(); + const connectionMap = new ConnectionMap(); const localTransport$ = new Subject(); - const connectionManagerData$ = new BehaviorSubject( - new Epoch(connectionManagerData), - ); + const connectionMap$ = new BehaviorSubject(new Epoch(connectionMap)); const publishers: Publisher[] = []; const publishing$ = new BehaviorSubject(false); @@ -522,9 +495,7 @@ describe("LocalMembership", () => { const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, - connectionManager: { - connectionManagerData$, - }, + connectionMap$: connectionMap$, localTransport$, }); @@ -542,14 +513,14 @@ describe("LocalMembership", () => { }, }); - const connectionManagerData2 = new ConnectionManagerData(); - connectionManagerData2.add( + const connectionMap2 = new ConnectionMap(); + connectionMap2.add( // clone because we will mutate this later. { ...connectionTransportAConnecting } as unknown as Connection, [], ); - connectionManagerData$.next(new Epoch(connectionManagerData2)); + connectionMap$.next(new Epoch(connectionMap2)); await flushPromises(); expect(localMembership.localMemberState$.value).toStrictEqual({ matrix: RTCMemberStatus.Connected, @@ -563,7 +534,7 @@ describe("LocalMembership", () => { }); ( - connectionManagerData2.getConnectionForTransport(mockTransport)! + connectionMap2.getConnectionForTransport(mockTransport)! .state$ as BehaviorSubject ).next({ state: "authenticated", @@ -659,8 +630,8 @@ describe("LocalMembership", () => { [boolean, HomeserverDisconnectReason | null] >([false, "membership"]); - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(connectionTransportAConnected, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(connectionTransportAConnected, []); createLocalMembership$({ scope, @@ -669,9 +640,7 @@ describe("LocalMembership", () => { combined$: hsReason$, rtsSession$: constant(RTCMemberStatus.Connected), }, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); @@ -696,8 +665,8 @@ describe("LocalMembership", () => { [boolean, HomeserverDisconnectReason | null] >([true, null]); - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(connectionTransportAConnected, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(connectionTransportAConnected, []); createLocalMembership$({ scope, @@ -706,9 +675,7 @@ describe("LocalMembership", () => { combined$: hsReason$, rtsSession$: constant(RTCMemberStatus.Connected), }, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); @@ -742,8 +709,8 @@ describe("LocalMembership", () => { state$: connectionState$, } as unknown as Connection; - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(mutableConnection, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(mutableConnection, []); createLocalMembership$({ scope, @@ -755,9 +722,7 @@ describe("LocalMembership", () => { ]), rtsSession$: constant(RTCMemberStatus.Connected), }, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); @@ -792,8 +757,8 @@ describe("LocalMembership", () => { [boolean, HomeserverDisconnectReason | null] >([true, null]); - const connectionManagerData = new ConnectionManagerData(); - connectionManagerData.add(connectionTransportAConnected, []); + const connectionMap = new ConnectionMap(); + connectionMap.add(connectionTransportAConnected, []); createLocalMembership$({ scope, @@ -802,9 +767,7 @@ describe("LocalMembership", () => { combined$: hsReason$, rtsSession$: constant(RTCMemberStatus.Connected), }, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); @@ -863,14 +826,12 @@ describe("LocalMembership", () => { localMembership: ReturnType; } => { const scope = new ObservableScope(); - const connectionManagerData = new ConnectionManagerData(); - if (connection) connectionManagerData.add(connection, []); + const connectionMap = new ConnectionMap(); + if (connection) connectionMap.add(connection, []); const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, - connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), - }, + connectionMap$: constant(new Epoch(connectionMap)), localTransport$: constant(mockTransport), }); return { scope, localMembership }; diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 2fb6e0229..482627ee2 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -47,8 +47,8 @@ import { deepCompare } from "matrix-js-sdk/lib/utils"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type Behavior } from "../../Behavior.ts"; -import { type IConnectionManager } from "../remoteMembers/ConnectionManager.ts"; -import { type ObservableScope } from "../../ObservableScope.ts"; +import { type ConnectionMap } from "../remoteMembers/ConnectionMap.ts"; +import { type Epoch, type ObservableScope } from "../../ObservableScope.ts"; import { type Publisher } from "./Publisher.ts"; import { type MuteStates } from "../../MuteStates.ts"; import { @@ -138,7 +138,7 @@ interface Props { // that the inputs for those createSomething$() functions should NOT contain any js-sdk objectes scope: ObservableScope; muteStates: MuteStates; - connectionManager: IConnectionManager; + connectionMap$: Behavior>; createPublisherFactory: (connection: Connection) => Publisher; joinMatrixRTC: ( transport: LivekitTransport, @@ -169,7 +169,7 @@ interface Props { * - * @param props The properties required to create the local membership. * @param props.scope The observable scope to use. - * @param props.connectionManager The connection manager to get connections from. + * @param props.connectionMap$ The connection map to get connections from. * @param props.createPublisherFactory Factory to create a publisher once we have a connection. * @param props.joinMatrixRTC Callback to join the matrix RTC session once we have a transport. * @param props.homeserverConnected The homeserver connected state. @@ -191,7 +191,7 @@ interface Props { */ export const createLocalMembership$ = ({ scope, - connectionManager, + connectionMap$, localTransport$: localTransportWithErrors$, homeserverConnected, createPublisherFactory, @@ -324,13 +324,11 @@ export const createLocalMembership$ = ({ // Drop Epoch data here since we will not combine this anymore const localConnection$ = scope.behavior( - combineLatest([ - connectionManager.connectionManagerData$, - localTransport$, - ]).pipe( - map(([{ value: connectionData }, transport]) => - connectionData.getConnectionForTransport(transport), - ), + combineLatest( + [connectionMap$, localTransport$], + ({ value: connectionMap }, transport) => + connectionMap.getConnectionForTransport(transport), + ).pipe( tap((connection) => { logger.info( `Local connection updated: ${JSON.stringify(connection?.transport)}`, diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 21e79ef97..5afd79dec 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -418,8 +418,8 @@ describe("Start connection states", () => { pendingConnect.reject(new Error("Client initiated disconnect")); await stopping; - // start() resolves rather than rejecting (it is not awaited by the - // ConnectionManager, so a rejection would be unhandled). + // start() resolves rather than rejecting (it is not awaited by the caller, + // so a rejection would be unhandled). await expect(started).resolves.toBeUndefined(); expect(capturedStates.at(-1)).toEqual({ state: "stopped" }); expect(capturedStates.some((st) => st instanceof Error)).toBe(false); diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index 35c2e2c75..15b0559cb 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.ts @@ -356,7 +356,7 @@ export class Connection { // stop() was called while we were connecting, which makes the pending // connect reject. That is the abort we asked for, not a failure, so // don't record an error state on a stopped connection or rethrow it - // (start() is not awaited by the ConnectionManager, so a throw here + // (start() is not necessarily awaited by the caller, so a throw here // becomes an unhandled promise rejection). this.logger.debug(`Connect aborted because the connection was stopped`); return; diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts deleted file mode 100644 index 3aa53d449..000000000 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ /dev/null @@ -1,269 +0,0 @@ -/* -Copyright 2025 Element Creations Ltd. -Copyright 2025 New Vector Ltd. - -SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial -Please see LICENSE in the repository root for full details. -*/ - -import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; -import { - combineLatest, - map, - type Observable, - of, - switchMap, - startWith, - catchError, - NEVER, -} 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"; - -import { type Behavior } from "../../Behavior.ts"; -import { type Connection } from "./Connection.ts"; -import { Epoch, type ObservableScope } from "../../ObservableScope.ts"; -import { generateItemsWithEpoch } from "../../../utils/observable.ts"; -import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; -import { type ConnectionFactory } from "./ConnectionFactory.ts"; - -export class ConnectionManagerData { - private readonly store: Map< - string, - { connection: Connection; participants: RemoteParticipant[] } - > = new Map(); - - public constructor(private readonly logger?: Logger) {} - - public add(connection: Connection, participants: RemoteParticipant[]): void { - const key = this.getKey(connection.transport); - const existing = this.store.get(key); - if (!existing) { - this.store.set(key, { connection, participants }); - } else { - // Transports are deduplicated by URL upstream, so this should never - // happen; if it does, members may be matched against the wrong room. - this.logger?.warn( - `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`, - ); - existing.participants.push(...participants); - } - } - - private getKey(transport: LivekitTransport): string { - // This is enough as a key because the ConnectionManager is already scoped by room. - // We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`. - return transport.livekit_service_url; - } - - public getConnections(): Connection[] { - return Array.from(this.store.values()).map(({ connection }) => connection); - } - - public getConnectionForTransport( - transport: LivekitTransport, - ): Connection | null { - return this.store.get(this.getKey(transport))?.connection ?? null; - } - - public getParticipantsForTransport( - transport: LivekitTransport, - ): RemoteParticipant[] { - const key = this.getKey(transport); - const existing = this.store.get(key); - if (existing) { - return existing.participants; - } - return []; - } -} - -interface Props { - scope: ObservableScope; - connectionFactory: ConnectionFactory; - localTransport$: Observable; - remoteTransports$: Behavior>; - - logger: Logger; - ownMembershipIdentity: CallMembershipIdentityParts; -} - -// TODO - write test for scopes (do we really need to bind scope) -export interface IConnectionManager { - connectionManagerData$: Behavior>; -} - -/** - * Crete a `ConnectionManager` - * @param props - Configuration object - * @param props.scope - The observable scope used by this object - * @param props.connectionFactory - Used to create new connections - * @param props.localTransport$ - The transport to publish local media on. (deduplicated with remoteTransports$) - * @param props.remoteTransports$ - All other transports. The connection manager will create connections for each transport. (deduplicated with localTransport$) - * @param props.ownMembershipIdentity - The own membership identity to use. - * @param props.logger - The logger to use. - - * - * Each of these behaviors can be interpreted as subscribed list of transports. - * - * Using `registerTransports` independent external modules can control what connections - * are created by the ConnectionManager. - * - * The connection manager will remove all duplicate transports in each subscibed list. - * - * See `unregisterAllTransports` and `unregisterTransport` for details on how to unsubscribe. - */ -export function createConnectionManager$({ - scope, - connectionFactory, - localTransport$, - remoteTransports$, - logger: parentLogger, - ownMembershipIdentity, -}: Props): IConnectionManager { - 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. - * - * This list does not include duplicate transports. - * - * It is build based on the list of subscribed transports (`transportsSubscriptions$`). - * externally this is modified via `registerTransports()`. - */ - const localAndRemoteTransports$: Behavior< - Epoch<{ transport: LivekitTransport; role: "publisher" | "subscriber" }[]> - > = scope.behavior( - combineLatest([localTransportAsArray$, remoteTransports$]).pipe( - // Combine local and remote transports into one transport array, tagged - // with the role we take on each of them - map(([localTransportAsArray, remoteTransports]) => { - const dedupedRemote = removeDuplicateTransports(remoteTransports.value); - const remoteWithoutLocal = dedupedRemote.filter( - (transport) => - !localTransportAsArray.find((l) => - areLivekitTransportsEqual(l, transport), - ), - ); - logger.debug( - "remoteWithoutLocal", - remoteWithoutLocal, - "localTransportAsArray", - localTransportAsArray, - ); - return new Epoch( - [ - ...localTransportAsArray.map((transport) => ({ - transport, - role: "publisher" as const, - })), - ...remoteWithoutLocal.map((transport) => ({ - transport, - role: "subscriber" as const, - })), - ], - remoteTransports.epoch, - ); - }), - ), - ); - - /** - * Connections for each transport in use by one or more session members. - */ - const connections$ = scope.behavior( - localAndRemoteTransports$.pipe( - generateItemsWithEpoch( - "ConnectionManager connections$", - function* (transports) { - for (const { transport, role } of transports) { - yield { - keys: [role, transport.livekit_service_url], - data: undefined, - }; - } - }, - (scope, _data$, role, serviceUrl) => { - const connection = connectionFactory.createConnection( - scope, - role, - { - type: "livekit", - livekit_service_url: serviceUrl, - }, - ownMembershipIdentity, - logger, - ); - // Start the connection immediately - // Use connection state to track connection progress - void connection.start(); - // TODO subscribe to connection state to retry or log issues? - return connection; - }, - ), - ), - ); - - const connectionManagerData$ = scope.behavior( - connections$.pipe( - switchMap((connections) => { - const epoch = connections.epoch; - - // Map the connections to list of {connection, participants}[] - const listOfConnectionsWithRemoteParticipants = connections.value.map( - (connection) => { - return connection.remoteParticipants$.pipe( - map((participants) => ({ - connection, - participants, - })), - ); - }, - ); - - // probably not required - - if (listOfConnectionsWithRemoteParticipants.length === 0) { - return of(new Epoch(new ConnectionManagerData(), epoch)); - } - - // combineLatest the several streams into a single stream with the ConnectionManagerData - return combineLatest(listOfConnectionsWithRemoteParticipants).pipe( - map( - (lists) => - new Epoch( - lists.reduce((data, { connection, participants }) => { - data.add(connection, participants); - return data; - }, new ConnectionManagerData(logger)), - epoch, - ), - ), - ); - }), - ), - new Epoch(new ConnectionManagerData(), -1), - ); - - return { connectionManagerData$ }; -} - -function removeDuplicateTransports( - transports: T[], -): T[] { - return transports.reduce((acc, transport) => { - if (!acc.some((t) => areLivekitTransportsEqual(t, transport))) - acc.push(transport); - return acc; - }, [] as T[]); -} diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts similarity index 85% rename from src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts rename to src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts index 89772fcf4..e0f7add1c 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts @@ -6,16 +6,13 @@ 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 LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; import { type RemoteParticipant } from "livekit-client"; import { logger } from "matrix-js-sdk/lib/logger"; import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts"; -import { - createConnectionManager$, - ConnectionManagerData, -} from "./ConnectionManager.ts"; +import { createConnectionMap$, ConnectionMap } from "./ConnectionMap.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { type Connection } from "./Connection.ts"; import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts"; @@ -34,6 +31,15 @@ const TRANSPORT_2: LivekitTransport = { livekit_service_url: "https://lk.sample.com", }; +function findConnection( + transport: LivekitTransport, + connections: Connection[], +): Connection | undefined { + return connections.find((c) => + areLivekitTransportsEqual(c.transport, transport), + ); +} + let fakeConnectionFactory: ConnectionFactory; let testScope: ObservableScope; @@ -71,10 +77,10 @@ afterEach(() => { describe("connections$ stream", () => { test("Should create and start new connections for each transports", () => { withTestScheduler(({ behavior, expectObservable }) => { - const { connectionManagerData$ } = createConnectionManager$({ + const connectionMap$ = createConnectionMap$({ scope: testScope, connectionFactory: fakeConnectionFactory, - localTransport$: NEVER, + localTransport: TRANSPORT_1, remoteTransports$: behavior("a", { a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0), }), @@ -83,7 +89,7 @@ describe("connections$ stream", () => { }); expectObservable( - connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())), + connectionMap$.pipe(mapEpoch((d) => d.getConnections())), ).toBe("a", { a: expect.toSatisfy((e: Epoch) => { const connections = e.value; @@ -93,15 +99,11 @@ describe("connections$ stream", () => { vi.mocked(fakeConnectionFactory).createConnection, ).toHaveBeenCalledTimes(2); - const conn1 = connections.find((c) => - areLivekitTransportsEqual(c.transport, TRANSPORT_1), - ); + const conn1 = findConnection(TRANSPORT_1, connections); expect(conn1).toBeDefined(); expect(conn1!.start).toHaveBeenCalled(); - const conn2 = connections.find((c) => - areLivekitTransportsEqual(c.transport, TRANSPORT_2), - ); + const conn2 = findConnection(TRANSPORT_2, connections); expect(conn2).toBeDefined(); expect(conn2!.start).toHaveBeenCalled(); return true; @@ -112,10 +114,10 @@ describe("connections$ stream", () => { test("Should start connection only once", () => { withTestScheduler(({ behavior, expectObservable }) => { - const { connectionManagerData$ } = createConnectionManager$({ + const connectionMap$ = createConnectionMap$({ scope: testScope, connectionFactory: fakeConnectionFactory, - localTransport$: NEVER, + localTransport: TRANSPORT_1, remoteTransports$: behavior("abcdef", { a: new Epoch([TRANSPORT_1], 0), b: new Epoch([TRANSPORT_1], 1), @@ -129,7 +131,7 @@ describe("connections$ stream", () => { }); expectObservable( - connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())), + connectionMap$.pipe(mapEpoch((d) => d.getConnections())), ).toBe("xxxxxa", { x: expect.anything(), a: expect.toSatisfy((e: Epoch) => { @@ -140,14 +142,10 @@ describe("connections$ stream", () => { vi.mocked(fakeConnectionFactory).createConnection, ).toHaveBeenCalledTimes(2); - const conn2 = connections.find((c) => - areLivekitTransportsEqual(c.transport, TRANSPORT_2), - ); + const conn2 = findConnection(TRANSPORT_2, connections); expect(conn2).toBeDefined(); - const conn1 = connections.find((c) => - areLivekitTransportsEqual(c.transport, TRANSPORT_1), - ); + const conn1 = findConnection(TRANSPORT_1, connections); expect(conn1).toBeDefined(); expect(conn1!.start).toHaveBeenCalledOnce(); @@ -159,10 +157,10 @@ describe("connections$ stream", () => { test("Should cleanup connections when not needed anymore", () => { withTestScheduler(({ behavior, expectObservable }) => { - const { connectionManagerData$ } = createConnectionManager$({ + const connectionMap$ = createConnectionMap$({ scope: testScope, connectionFactory: fakeConnectionFactory, - localTransport$: NEVER, + localTransport: TRANSPORT_1, remoteTransports$: behavior("abc", { a: new Epoch([TRANSPORT_1], 0), b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1), @@ -173,7 +171,7 @@ describe("connections$ stream", () => { }); expectObservable( - connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())), + connectionMap$.pipe(mapEpoch((d) => d.getConnections())), ).toBe("xab", { x: expect.anything(), a: expect.toSatisfy((e: Epoch) => { @@ -186,8 +184,9 @@ describe("connections$ stream", () => { expect(connections.length).toBe(1); // The second connection should have been stopped has it is no longer needed. - const connection2 = allCreatedConnections.find((c) => - areLivekitTransportsEqual(c.transport, TRANSPORT_2), + const connection2 = findConnection( + TRANSPORT_2, + allCreatedConnections, ); expect(connection2).toBeDefined(); expect(connection2!.stop).toHaveBeenCalled(); @@ -203,10 +202,10 @@ describe("connections$ stream", () => { }); }); -describe("ConnectionManagerData", () => { +describe("ConnectionMap", () => { test("warns when a second connection to the same URL is merged", () => { const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); - const data = new ConnectionManagerData(logger); + const data = new ConnectionMap(logger); const connection = { transport: TRANSPORT_1 } as unknown as Connection; const p = (identity: string): RemoteParticipant => ({ identity }) as unknown as RemoteParticipant; @@ -221,7 +220,7 @@ describe("ConnectionManagerData", () => { }); }); -describe("connectionManagerData$ stream", () => { +describe("connectionMap$", () => { // Used in test to control fake connections' remoteParticipants$ streams let fakeRemoteParticipantsStreams: Map>; @@ -298,10 +297,10 @@ describe("connectionManagerData$ stream", () => { ); // ============================== - const { connectionManagerData$ } = createConnectionManager$({ + const connectionMap$ = createConnectionMap$({ scope: testScope, connectionFactory: fakeConnectionFactory, - localTransport$: NEVER, + localTransport: TRANSPORT_1, remoteTransports$: behavior("a", { a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0), }), @@ -309,16 +308,16 @@ describe("connectionManagerData$ stream", () => { ownMembershipIdentity: ownMemberMock, }); - expectObservable(connectionManagerData$).toBe("abcd", { + expectObservable(connectionMap$).toBe("abcd", { a: expect.toSatisfy((e) => { - const data: ConnectionManagerData = e.value; + const data: ConnectionMap = e.value; expect(data.getConnections().length).toBe(2); expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(0); expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0); return true; }), b: expect.toSatisfy((e) => { - const data: ConnectionManagerData = e.value; + const data: ConnectionMap = e.value; expect(data.getConnections().length).toBe(2); expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1); expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0); @@ -328,7 +327,7 @@ describe("connectionManagerData$ stream", () => { return true; }), c: expect.toSatisfy((e) => { - const data: ConnectionManagerData = e.value; + const data: ConnectionMap = e.value; expect(data.getConnections().length).toBe(2); expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1); expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1); @@ -341,7 +340,7 @@ describe("connectionManagerData$ stream", () => { return true; }), d: expect.toSatisfy((e) => { - const data: ConnectionManagerData = e.value; + const data: ConnectionMap = e.value; expect(data.getConnections().length).toBe(2); expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(2); expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1); diff --git a/src/state/CallViewModel/remoteMembers/ConnectionMap.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.ts new file mode 100644 index 000000000..67253bf38 --- /dev/null +++ b/src/state/CallViewModel/remoteMembers/ConnectionMap.ts @@ -0,0 +1,193 @@ +/* +Copyright 2025-2026 Element Creations Ltd. +Copyright 2025 New Vector Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; +import { combineLatest, map, of, switchMap, tap } 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"; + +import { type Behavior } from "../../Behavior.ts"; +import { type Connection } from "./Connection.ts"; +import { + Epoch, + mapEpoch, + type ObservableScope, +} from "../../ObservableScope.ts"; +import { generateItemsWithEpoch } from "../../../utils/observable.ts"; +import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; +import { type ConnectionFactory } from "./ConnectionFactory.ts"; + +function removeDuplicateTransports( + transports: T[], +): T[] { + return transports.reduce((acc, transport) => { + if (!acc.some((t) => areLivekitTransportsEqual(t, transport))) + acc.push(transport); + return acc; + }, [] as T[]); +} + +export class ConnectionMap { + private readonly map = new Map< + string, + { connection: Connection; participants: RemoteParticipant[] } + >(); + + public constructor(private readonly logger?: Logger) {} + + public add(connection: Connection, participants: RemoteParticipant[]): void { + const key = this.getKey(connection.transport); + const existing = this.map.get(key); + if (!existing) { + this.map.set(key, { connection, participants }); + } else { + // Transports are deduplicated by URL upstream, so this should never + // happen; if it does, members may be matched against the wrong room. + this.logger?.warn( + `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`, + ); + existing.participants.push(...participants); + } + } + + private getKey(transport: LivekitTransport): string { + // This is enough as a key because the ConnectionMap is already scoped by room. + // We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`. + return transport.livekit_service_url; + } + + public getConnections(): Connection[] { + return Array.from(this.map.values()).map(({ connection }) => connection); + } + + public getConnectionForTransport( + transport: LivekitTransport, + ): Connection | null { + return this.map.get(this.getKey(transport))?.connection ?? null; + } + + public getParticipantsForTransport( + transport: LivekitTransport, + ): RemoteParticipant[] { + const existing = this.map.get(this.getKey(transport)); + if (existing) { + return existing.participants; + } + return []; + } +} + +interface Props { + scope: ObservableScope; + connectionFactory: ConnectionFactory; + localTransport: LivekitTransport; + remoteTransports$: Behavior>; + + logger: Logger; + ownMembershipIdentity: CallMembershipIdentityParts; +} + +/** + * Create a dynamic map of connections and their associated backend participants. + * @param props - Configuration object + * @param props.scope - The observable scope used by this object + * @param props.connectionFactory - Used to create new connections + * @param props.localTransport - The transport to publish local media on. (deduplicated with remoteTransports$) + * @param props.remoteTransports$ - All other transports. The connection manager will create connections for each transport. (deduplicated with localTransport) + * @param props.ownMembershipIdentity - The own membership identity to use. + * @param props.logger - The logger to use. + */ +export function createConnectionMap$({ + scope, + connectionFactory, + localTransport, + remoteTransports$, + logger: parentLogger, + ownMembershipIdentity, +}: Props): Behavior> { + const logger = parentLogger.getChild("[ConnectionMap]"); + // TODO logger: only construct one logger from the client and make it compatible via a EC specific sing + + /** + * All transports currently represented in the connection map. This list does + * not include duplicate transports. + */ + const localAndRemoteTransports$: Behavior> = + scope.behavior( + remoteTransports$.pipe( + mapEpoch((remoteTransports) => + removeDuplicateTransports([localTransport, ...remoteTransports]), + ), + tap((transports) => + logger.debug("localAndRemoteTransports$ = ", transports), + ), + ), + ); + + /** + * Connections for each transport in use by one or more session members. + */ + const connections$ = scope.behavior( + localAndRemoteTransports$.pipe( + generateItemsWithEpoch( + "ConnectionMap connections$", + function* (transports) { + for (const transport of transports) { + const role = areLivekitTransportsEqual(transport, localTransport) + ? ("publisher" as const) + : ("subscriber" as const); + yield { + keys: [role, transport.livekit_service_url], + data: undefined, + }; + } + }, + (scope, _data$, role, serviceUrl) => { + const connection = connectionFactory.createConnection( + scope, + role, + { type: "livekit", livekit_service_url: serviceUrl }, + ownMembershipIdentity, + logger, + ); + // Start the connection immediately + // Use connection state to track connection progress + void connection.start(); + // TODO subscribe to connection state to retry or log issues? + return connection; + }, + ), + ), + ); + + return scope.behavior( + connections$.pipe( + switchMap(({ value: connections, epoch }) => { + if (connections.length === 0) + return of(new Epoch(new ConnectionMap(), epoch)); + + return combineLatest( + // Map the connections to list of Observable<{connection, participants}> + connections.map((connection) => + connection.remoteParticipants$.pipe( + map((participants) => ({ connection, participants })), + ), + ), + // Collect into a single ConnectionMap + (...list) => { + const map = new ConnectionMap(logger); + for (const item of list) + map.add(item.connection, item.participants); + return new Epoch(map, epoch); + }, + ); + }), + ), + ); +} diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts index 2e0051928..42548a79a 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts @@ -13,7 +13,6 @@ import { } from "matrix-js-sdk/lib/matrixrtc"; import { BehaviorSubject, combineLatest, map, type Observable } from "rxjs"; -import { type IConnectionManager } from "./ConnectionManager.ts"; import { type RemoteMatrixLivekitMember, createRemoteMatrixLivekitMembers$, @@ -24,7 +23,7 @@ import { ObservableScope, trackEpoch, } from "../../ObservableScope.ts"; -import { ConnectionManagerData } from "./ConnectionManager.ts"; +import { ConnectionMap } from "./ConnectionMap.ts"; import { flushPromises, mockRtcMembership, @@ -78,23 +77,18 @@ function epochMeWith$( test("should signal participant not yet connected to livekit", async () => { const info = vi.spyOn(logger, "info"); const mockedMemberships$ = new BehaviorSubject([bobMembership]); - const mockConnectionManagerData$ = new BehaviorSubject( - new ConnectionManagerData(), - ); + const mockConnectionMap$ = new BehaviorSubject(new ConnectionMap()); const { memberships$, membershipsWithTransport$ } = createEpochedMemberships$(mockedMemberships$); - const connectionManagerData$ = epochMeWith$( - memberships$, - mockConnectionManagerData$, + const connectionMap$ = testScope.behavior( + epochMeWith$(memberships$, mockConnectionMap$), ); const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({ scope: testScope, membershipsWithTransport$: testScope.behavior(membershipsWithTransport$), - connectionManager: { - connectionManagerData$: connectionManagerData$, - } as unknown as IConnectionManager, + connectionMap$, localUser: localRtcMember, }); @@ -152,22 +146,19 @@ test("should signal participant on a connection that is publishing", async () => const connection = { transport: bobMembership.getTransport(), } as unknown as Connection; - const dataWithPublisher = new ConnectionManagerData(); + const dataWithPublisher = new ConnectionMap(); dataWithPublisher.add(connection, [ mockRemoteParticipant({ identity: bobParticipantId }), ]); - const connectionManagerData$ = epochMeWith$( - memberships$, - constant(dataWithPublisher), + const connectionMap$ = testScope.behavior( + epochMeWith$(memberships$, constant(dataWithPublisher)), ); const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({ scope: testScope, membershipsWithTransport$: testScope.behavior(membershipsWithTransport$), - connectionManager: { - connectionManagerData$: connectionManagerData$, - } as unknown as IConnectionManager, + connectionMap$, localUser: localRtcMember, }); @@ -195,20 +186,17 @@ test("should signal participant on a connection that is not publishing", async ( const connection = { transport: bobMembership.getTransport(), } as unknown as Connection; - const dataWithPublisher = new ConnectionManagerData(); + const dataWithPublisher = new ConnectionMap(); dataWithPublisher.add(connection, []); - const connectionManagerData$ = epochMeWith$( - memberships$, - constant(dataWithPublisher), + const connectionMap$ = testScope.behavior( + epochMeWith$(memberships$, constant(dataWithPublisher)), ); const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({ scope: testScope, membershipsWithTransport$: testScope.behavior(membershipsWithTransport$), - connectionManager: { - connectionManagerData$: connectionManagerData$, - } as unknown as IConnectionManager, + connectionMap$, localUser: localRtcMember, }); await flushPromises(); @@ -228,7 +216,7 @@ describe("Publication edge case", () => { const { memberships$, membershipsWithTransport$ } = createEpochedMemberships$(constant([bobMembership, carlMembership])); - const connectionWithPublisher = new ConnectionManagerData(); + const connectionWithPublisher = new ConnectionMap(); const bobParticipantId = fallbackMemberId( bobMembership.userId, bobMembership.deviceId, @@ -247,17 +235,14 @@ describe("Publication edge case", () => { mockRemoteParticipant({ identity: bobParticipantId }), ]); - const connectionManagerData$ = epochMeWith$( - memberships$, - constant(connectionWithPublisher), + const connectionMap$ = testScope.behavior( + epochMeWith$(memberships$, constant(connectionWithPublisher)), ); const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({ scope: testScope, membershipsWithTransport$: testScope.behavior(membershipsWithTransport$), - connectionManager: { - connectionManagerData$: connectionManagerData$, - } as unknown as IConnectionManager, + connectionMap$, localUser: localRtcMember, }); await flushPromises(); @@ -287,7 +272,7 @@ test("bob is publishing in the wrong connection", async () => { const { memberships$, membershipsWithTransport$ } = createEpochedMemberships$(mockedMemberships$); - const connectionWithPublisher = new ConnectionManagerData(); + const connectionWithPublisher = new ConnectionMap(); const bobParticipantId = fallbackMemberId( bobMembership.userId, @@ -306,17 +291,14 @@ test("bob is publishing in the wrong connection", async () => { const connectionsWithPublisher$ = new BehaviorSubject( connectionWithPublisher, ); - const connectionManagerData$ = epochMeWith$( - memberships$, - connectionsWithPublisher$, + const connectionMap$ = testScope.behavior( + epochMeWith$(memberships$, connectionsWithPublisher$), ); const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({ scope: testScope, membershipsWithTransport$: testScope.behavior(membershipsWithTransport$), - connectionManager: { - connectionManagerData$: connectionManagerData$, - } as unknown as IConnectionManager, + connectionMap$, localUser: localRtcMember, }); diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts index d899fd9e2..8c5f21d77 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts @@ -14,7 +14,7 @@ import { combineLatest, filter, map } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; import { type Behavior } from "../../Behavior"; -import { type IConnectionManager } from "./ConnectionManager"; +import { type ConnectionMap } from "./ConnectionMap"; import { Epoch, type ObservableScope } from "../../ObservableScope"; import { type Connection } from "./Connection"; import { generateItemsWithEpoch } from "../../../utils/observable"; @@ -62,7 +62,7 @@ interface Props { membershipsWithTransport$: Behavior< Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]> >; - connectionManager: IConnectionManager; + connectionMap$: Behavior>; localUser: { deviceId: string; userId: string }; } @@ -72,24 +72,21 @@ interface Props { * It has a small public interface: * - in (via constructor): * - an observable of CallMembership[] to track the call members (The matrix side) - * - a `ConnectionManager` for the lk rooms (The livekit side) + * - a `ConnectionMap` for the lk rooms (The livekit side) * - out (via public Observable): * - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data. */ export function createRemoteMatrixLivekitMembers$({ scope, membershipsWithTransport$, - connectionManager, + connectionMap$, localUser, }: Props): Behavior> { /** * Behavior of all the remote call members and their associated livekit data (if available). */ return scope.behavior( - combineLatest([ - membershipsWithTransport$, - connectionManager.connectionManagerData$, - ]).pipe( + combineLatest([membershipsWithTransport$, connectionMap$]).pipe( filter((values) => values.every((value) => value.epoch === values[0].epoch), ), diff --git a/src/state/CallViewModel/remoteMembers/integration.test.ts b/src/state/CallViewModel/remoteMembers/integration.test.ts index 07b8d6f58..df3ff8dc3 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, NEVER } from "rxjs"; +import { BehaviorSubject } from "rxjs"; import { type Room as LivekitRoom } from "livekit-client"; import EventEmitter from "events"; import fetchMock from "fetch-mock"; @@ -21,6 +21,7 @@ import { import { ECConnectionFactory } from "./ConnectionFactory.ts"; import { type ClientOpenIDParts } from "../../../livekit/auth"; import { + exampleTransport, mockMediaDevices, mockRtcMembership, ownMemberMock, @@ -32,11 +33,11 @@ import { createRemoteMatrixLivekitMembers$, type RemoteMatrixLivekitMember, } from "./MatrixLivekitMembers.ts"; -import { createConnectionManager$ } from "./ConnectionManager.ts"; +import { createConnectionMap$ } from "./ConnectionMap.ts"; import { membershipsAndTransports$ } from "../../SessionBehaviors.ts"; import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts"; -// Test the integration of ConnectionManager and MatrixLivekitMerger +// Test the integration of ConnectionMap and MatrixLivekitMembers let testScope: ObservableScope; let ecConnectionFactory: ECConnectionFactory; @@ -122,10 +123,10 @@ test("bob, carl, then bob joining no tracks yet", () => { memberships$, ); - const connectionManager = createConnectionManager$({ + const connectionMap$ = createConnectionMap$({ scope: testScope, connectionFactory: ecConnectionFactory, - localTransport$: NEVER, + localTransport: exampleTransport, remoteTransports$: membershipsAndTransports.transports$, logger: logger, ownMembershipIdentity: ownMemberMock, @@ -135,7 +136,7 @@ test("bob, carl, then bob joining no tracks yet", () => { scope: testScope, membershipsWithTransport$: membershipsAndTransports.membershipsWithTransport$, - connectionManager, + connectionMap$, localUser: localRtcMember, }); diff --git a/src/utils/observable.ts b/src/utils/observable.ts index ca51ad196..b35e523d6 100644 --- a/src/utils/observable.ts +++ b/src/utils/observable.ts @@ -218,6 +218,28 @@ export function filterBehavior( ); } +/** + * Applies a given projection to each value emitted by the source Observable. + * The `project` function additionally receives an {@link ObservableScope}. + */ +export function mapScoped( + name: string, + project: (a: A, scope: ObservableScope) => B, +): OperatorFunction { + return (a$) => + a$.pipe( + generateItemsInternal( + name, + function* (a) { + yield { keys: [a], data: undefined }; + }, + (scope, _data$, a) => project(a, scope), + (items) => items, + ), + map(([item]) => item), + ); +} + function generateItemsInternal< Input, Keys extends [unknown, ...unknown[]],