Matrix/element-call-Github
1
0
Fork 0
mirror of https://github.com/vector-im/element-call.git synced 2026-09-22 22:29:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4273 from element-hq/simplify-local-transport

Browse Source 
Simplify local transport code
This commit is contained in:
Robin
2026-09-22 12:15:34 +02:00
committed by GitHub
parent 9f7c35cd8f c9f911b7c9
commit 1f84cb9c5c
12 changed files with 309 additions and 680 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/state/CallViewModel/CallViewModel.ts
+14 -24
View File
@@ -16,7 +16,6 @@ import {
import { type Room as MatrixRoom } from "matrix-js-sdk"; import { type Room as MatrixRoom } from "matrix-js-sdk";
import { import {
BehaviorSubject, BehaviorSubject,
catchError,
combineLatest, combineLatest,
distinctUntilChanged, distinctUntilChanged,
filter, filter,
@@ -39,6 +38,7 @@ import {
throttleTime, throttleTime,
timer, timer,
takeUntil, takeUntil,
from,
} from "rxjs"; } from "rxjs";
import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger"; import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { import {
@@ -112,7 +112,7 @@ import {
TransportState, TransportState,
} from "./localMember/LocalMember.ts"; } from "./localMember/LocalMember.ts";
import { import {
createLocalTransport$, getLocalTransport,
type LocalTransport, type LocalTransport,
} from "./localMember/LocalTransport.ts"; } from "./localMember/LocalTransport.ts";
import { import {
@@ -573,16 +573,16 @@ export function createCallViewModel$(
: `${userId}:${deviceId}`, : `${userId}:${deviceId}`,
}; };
const localTransport = const localTransport$ = options.localTransport
options.localTransport ?? ? constant(options.localTransport)
createLocalTransport$({ : from(
scope: scope, getLocalTransport({
memberships$: memberships$, ownMembershipIdentity,
ownMembershipIdentity, client,
client, roomId: matrixRoom.roomId,
roomId: matrixRoom.roomId, matrixRTCMode,
matrixRTCMode, }),
}); );
const connectionFactory = const connectionFactory =
options.connectionFactory ?? options.connectionFactory ??
@@ -599,17 +599,7 @@ export function createCallViewModel$(
const connectionManager = createConnectionManager$({ const connectionManager = createConnectionManager$({
scope: scope, scope: scope,
connectionFactory: connectionFactory, connectionFactory: connectionFactory,
localTransport$: scope.behavior( localTransport$,
localTransport.active$.pipe(
catchError((e: unknown) => {
logger.info(
"could not pass local transport to createConnectionManager$. localTransport$ threw an error",
e,
);
return of(null);
}),
),
),
remoteTransports$: membershipsAndTransports.transports$, remoteTransports$: membershipsAndTransports.transports$,
logger: logger, logger: logger,
ownMembershipIdentity, ownMembershipIdentity,
@@ -665,7 +655,7 @@ export function createCallViewModel$(
connectionManager, connectionManager,
client, client,
matrixRTCSession, matrixRTCSession,
localTransport, localTransport$,
roomId: matrixRoom.roomId, roomId: matrixRoom.roomId,
hideScreensharing, hideScreensharing,
hostBridge, hostBridge,
src/state/CallViewModel/CallViewModelTestUtils.ts
+2 -5
View File
@@ -210,11 +210,8 @@ export function withCallViewModel(mode: MatrixRTCMode) {
connectionState$, connectionState$,
windowSize$, windowSize$,
localTransport: { localTransport: {
active$: constant({ transport: exampleTransport,
transport: exampleTransport, sfuConfig: exampleSfuConfig,
sfuConfig: exampleSfuConfig,
}),
advertised$: constant(exampleTransport),
}, },
connectionFactory: { connectionFactory: {
createConnection( createConnection(
src/state/CallViewModel/localMember/LocalMember.test.ts
+30 -207
View File
@@ -21,7 +21,7 @@ import {
beforeEach, beforeEach,
afterEach, afterEach,
} from "vitest"; } from "vitest";
import { BehaviorSubject, map, of } from "rxjs"; import { BehaviorSubject, map, of, Subject } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger"; import { logger } from "matrix-js-sdk/lib/logger";
import { type LocalParticipant, type LocalTrack } from "livekit-client"; import { type LocalParticipant, type LocalTrack } from "livekit-client";
import fetchMock from "fetch-mock"; import fetchMock from "fetch-mock";
@@ -50,10 +50,7 @@ import {
TrackState, TrackState,
watchScreenShareToggle, watchScreenShareToggle,
} from "./LocalMember"; } from "./LocalMember";
import { import { MatrixRTCTransportMissingError } from "../../../utils/errors";
FailToGetOpenIdToken,
MatrixRTCTransportMissingError,
} from "../../../utils/errors";
import { Epoch, ObservableScope } from "../../ObservableScope"; import { Epoch, ObservableScope } from "../../ObservableScope";
import { constant } from "../../Behavior"; import { constant } from "../../Behavior";
import { ConnectionManagerData } from "../remoteMembers/ConnectionManager"; import { ConnectionManagerData } from "../remoteMembers/ConnectionManager";
@@ -61,10 +58,7 @@ import { ConnectionState, type Connection } from "../remoteMembers/Connection";
import { type Publisher } from "./Publisher"; import { type Publisher } from "./Publisher";
import { initializeWidget } from "../../../widget"; import { initializeWidget } from "../../../widget";
import { nullHostBridge } from "../../../HostBridge"; import { nullHostBridge } from "../../../HostBridge";
import { import { type LocalTransport } from "./LocalTransport";
type LocalTransport,
type LocalTransportWithSFUConfig,
} from "./LocalTransport";
import * as openIDSFU from "../../../livekit/openIDSFU"; import * as openIDSFU from "../../../livekit/openIDSFU";
initializeWidget(); initializeWidget();
@@ -261,7 +255,7 @@ describe("LocalMembership", () => {
}); });
it("throws error on missing RTC config error", () => { it("throws error on missing RTC config error", () => {
withTestScheduler(({ scope, hot, behavior, expectObservable }) => { withTestScheduler(({ scope, hot, expectObservable }) => {
const localTransport$ = scope.behavior<null | LivekitTransportConfig>( const localTransport$ = scope.behavior<null | LivekitTransportConfig>(
hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")), hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")),
null, null,
@@ -277,16 +271,15 @@ describe("LocalMembership", () => {
), ),
}; };
const aLocalTransport: LocalTransport = {
advertised$: localTransport$,
active$: behavior("a", { a: null }),
};
const localMembership = createLocalMembership$({ const localMembership = createLocalMembership$({
scope, scope,
...defaultCreateLocalMemberValues, ...defaultCreateLocalMemberValues,
connectionManager: mockConnectionManager, connectionManager: mockConnectionManager,
localTransport: aLocalTransport, localTransport$: hot(
"1ms #",
{},
new MatrixRTCTransportMissingError("domain.com"),
),
}); });
expectObservable(localMembership.localMemberState$).toBe("ne", { 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<null | LivekitTransportConfig>(
hot("--#", {}, new FailToGetOpenIdToken(new Error("foo"))),
null,
);
// Populate a connection for active
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportBConnected, []);
const mockConnectionManager = {
transports$: constant(new Epoch([bTransport])),
connectionManagerData$: constant(new Epoch(connectionManagerData)),
};
const aLocalTransport: LocalTransport = {
advertised$,
active$: behavior("a", { n: null, a: bTransportWithSFUConfig }),
};
defaultCreateLocalMemberValues.createPublisherFactory.mockImplementation(
() => {
return {} as unknown as Publisher;
},
);
const publisherFactory =
defaultCreateLocalMemberValues.createPublisherFactory as ReturnType<
typeof vi.fn
>;
const localMembership = createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
connectionManager: mockConnectionManager,
localTransport: aLocalTransport,
});
expectObservable(localMembership.localMemberState$).toBe("n-e", {
n: TransportState.Waiting,
e: expect.toSatisfy((e) => e instanceof FailToGetOpenIdToken),
});
// Should not have created any publisher
expect(publisherFactory).toHaveBeenCalledTimes(0);
});
});
it("logs if callIntent cannot be updated", async () => { it("logs if callIntent cannot be updated", async () => {
const scope = new ObservableScope(); const scope = new ObservableScope();
const aLocalTransport: LocalTransport = {
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
};
const mockConnectionManager = { const mockConnectionManager = {
transports$: constant(new Epoch([])), transports$: constant(new Epoch([])),
connectionManagerData$: constant(new Epoch(new ConnectionManagerData())), connectionManagerData$: constant(new Epoch(new ConnectionManagerData())),
@@ -366,7 +307,7 @@ describe("LocalMembership", () => {
leaveRoomSession: vi.fn(), leaveRoomSession: vi.fn(),
}, },
connectionManager: mockConnectionManager, connectionManager: mockConnectionManager,
localTransport: aLocalTransport, localTransport$: constant(mockTransport),
}); });
const expextedLog = const expextedLog =
"'not connected yet' while updating the call intent (this is expected on startup)"; "'not connected yet' while updating the call intent (this is expected on startup)";
@@ -378,33 +319,19 @@ describe("LocalMembership", () => {
scope.end(); scope.end();
}); });
const aTransport = { const mockTransportConfig = {
livekit_service_url: "a", livekit_service_url: "a",
} as LivekitTransportConfig; } as LivekitTransportConfig;
const aTransportWithSFUConfig = { const mockTransport = {
transport: aTransport, transport: mockTransportConfig,
sfuConfig: { sfuConfig: {
jwt: "foo", jwt: "foo",
livekitAlias: "bar", livekitAlias: "bar",
livekitIdentity: "baz", livekitIdentity: "baz",
url: "bro", url: "bro",
}, },
} as LocalTransportWithSFUConfig; } as LocalTransport;
const bTransport = {
livekit_service_url: "b",
} as LivekitTransportConfig;
const bTransportWithSFUConfig = {
transport: bTransport,
sfuConfig: {
jwt: "foo2",
livekitAlias: "bar2",
livekitIdentity: "baz2",
url: "bro2",
},
} as LocalTransportWithSFUConfig;
const connectionTransportAConnected = { const connectionTransportAConnected = {
livekitRoom: mockLivekitRoom({ livekitRoom: mockLivekitRoom({
@@ -414,18 +341,13 @@ describe("LocalMembership", () => {
} as unknown as LocalParticipant, } as unknown as LocalParticipant,
}), }),
state$: constant(ConnectionState.LivekitConnected), state$: constant(ConnectionState.LivekitConnected),
transport: aTransport, transport: mockTransportConfig,
} as unknown as Connection; } as Connection;
const connectionTransportAConnecting = { const connectionTransportAConnecting = {
...connectionTransportAConnected, ...connectionTransportAConnected,
state$: constant(ConnectionState.LivekitConnecting), state$: constant(ConnectionState.LivekitConnecting),
livekitRoom: mockLivekitRoom({}), livekitRoom: mockLivekitRoom({}),
} as unknown as Connection; } as unknown as Connection;
const connectionTransportBConnected = {
state$: constant(ConnectionState.LivekitConnected),
transport: bTransport,
livekitRoom: mockLivekitRoom({}),
} as unknown as Connection;
const authCallSpy = vi const authCallSpy = vi
.spyOn(openIDSFU, "getSFUConfigWithOpenID") .spyOn(openIDSFU, "getSFUConfigWithOpenID")
@@ -459,10 +381,7 @@ describe("LocalMembership", () => {
), ),
}, },
joinMatrixRTC, joinMatrixRTC,
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
delayId$, delayId$,
}); });
@@ -471,7 +390,7 @@ describe("LocalMembership", () => {
await flushPromises(); await flushPromises();
// Joins with timings appropriate for the level of delegation support // Joins with timings appropriate for the level of delegation support
expect(joinMatrixRTC).toHaveBeenCalledWith( expect(joinMatrixRTC).toHaveBeenCalledWith(
aTransport, mockTransportConfig,
delayedLeaveTimings, delayedLeaveTimings,
); );
@@ -505,75 +424,7 @@ describe("LocalMembership", () => {
}, },
); );
it("recreates publisher if new connection is used, always unpublish and end tracks", async () => { it("only starts tracks if requested", async () => {
const scope = new ObservableScope();
const activeTransport$ = new BehaviorSubject(aTransportWithSFUConfig);
const aLocalTransport: LocalTransport = {
advertised$: constant(aTransport),
active$: activeTransport$,
};
const publishers: Publisher[] = [];
let seed = 0;
defaultCreateLocalMemberValues.createPublisherFactory.mockImplementation(
() => {
const a = seed;
seed += 1;
logger.info(`creating [${a}]`);
const p = {
// It is enought to check if destroy is called. Destroy itself is tested in the publisher to make sure it does
// all the cleanup we need.
destroy: vi.fn(),
stopPublishing: vi.fn().mockImplementation(() => {
logger.info(`stopPublishing [${a}]`);
}),
stopTracks: vi.fn(),
};
publishers.push(p as unknown as Publisher);
return p;
},
);
const publisherFactory =
defaultCreateLocalMemberValues.createPublisherFactory as ReturnType<
typeof vi.fn
>;
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportAConnected, []);
connectionManagerData.add(connectionTransportBConnected, []);
createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport: aLocalTransport,
});
await flushPromises();
activeTransport$.next({
...aTransportWithSFUConfig,
transport: bTransport,
});
await flushPromises();
expect(publisherFactory).toHaveBeenCalledTimes(2);
expect(publishers.length).toBe(2);
// stop the first Publisher and let the second one life.
expect(publishers[0].destroy).toHaveBeenCalled();
expect(publishers[1].destroy).not.toHaveBeenCalled();
expect(publisherFactory.mock.calls[0][0].transport).toBe(aTransport);
expect(publisherFactory.mock.calls[1][0].transport).toBe(bTransport);
scope.end();
await flushPromises();
// stop all tracks after ending scopes
expect(publishers[1].destroy).toHaveBeenCalled();
// expect(publishers[1].stopTracks).toHaveBeenCalled();
defaultCreateLocalMemberValues.createPublisherFactory.mockReset();
});
it("only start tracks if requested", async () => {
const scope = new ObservableScope(); const scope = new ObservableScope();
const publishers: Publisher[] = []; const publishers: Publisher[] = [];
@@ -602,11 +453,6 @@ describe("LocalMembership", () => {
typeof vi.fn typeof vi.fn
>; >;
const aLocalTransport: LocalTransport = {
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
};
const connectionManagerData = new ConnectionManagerData(); const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportAConnected, []); connectionManagerData.add(connectionTransportAConnected, []);
// connectionManagerData.add(connectionTransportB, []); // connectionManagerData.add(connectionTransportB, []);
@@ -616,7 +462,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: aLocalTransport, localTransport$: constant(mockTransport),
}); });
await flushPromises(); await flushPromises();
expect(publisherFactory).toHaveBeenCalledOnce(); expect(publisherFactory).toHaveBeenCalledOnce();
@@ -637,16 +483,8 @@ describe("LocalMembership", () => {
// //
it("tracks livekit state correctly", async () => { it("tracks livekit state correctly", async () => {
const scope = new ObservableScope(); const scope = new ObservableScope();
const connectionManagerData = new ConnectionManagerData(); const connectionManagerData = new ConnectionManagerData();
const localTransport$ = new Subject<LocalTransport>();
const activeTransport$ =
new BehaviorSubject<null | LocalTransportWithSFUConfig>(null);
const aLocalTransport: LocalTransport = {
advertised$: constant(aTransport),
active$: activeTransport$,
};
const connectionManagerData$ = new BehaviorSubject( const connectionManagerData$ = new BehaviorSubject(
new Epoch(connectionManagerData), new Epoch(connectionManagerData),
@@ -687,14 +525,14 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$, connectionManagerData$,
}, },
localTransport: aLocalTransport, localTransport$,
}); });
await flushPromises(); await flushPromises();
expect(localMembership.localMemberState$.value).toStrictEqual( expect(localMembership.localMemberState$.value).toStrictEqual(
TransportState.Waiting, TransportState.Waiting,
); );
activeTransport$.next(aTransportWithSFUConfig); localTransport$.next(mockTransport);
await flushPromises(); await flushPromises();
expect(localMembership.localMemberState$.value).toStrictEqual({ expect(localMembership.localMemberState$.value).toStrictEqual({
matrix: RTCMemberStatus.Connected, matrix: RTCMemberStatus.Connected,
@@ -719,7 +557,7 @@ describe("LocalMembership", () => {
}); });
( (
connectionManagerData2.getConnectionForTransport(aTransport)! connectionManagerData2.getConnectionForTransport(mockTransportConfig)!
.state$ as BehaviorSubject<ConnectionState> .state$ as BehaviorSubject<ConnectionState>
).next(ConnectionState.LivekitConnected); ).next(ConnectionState.LivekitConnected);
expect(localMembership.localMemberState$.value).toStrictEqual({ expect(localMembership.localMemberState$.value).toStrictEqual({
@@ -822,10 +660,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
}); });
await flushPromises(); await flushPromises();
@@ -862,10 +697,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
}); });
await flushPromises(); await flushPromises();
@@ -913,10 +745,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
}); });
await flushPromises(); await flushPromises();
@@ -957,10 +786,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
}); });
await flushPromises(); await flushPromises();
@@ -1026,10 +852,7 @@ describe("LocalMembership", () => {
connectionManager: { connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)), connectionManagerData$: constant(new Epoch(connectionManagerData)),
}, },
localTransport: { localTransport$: constant(mockTransport),
advertised$: constant(aTransport),
active$: constant(aTransportWithSFUConfig),
},
}); });
return { scope, localMembership }; return { scope, localMembership };
}; };
@@ -1039,7 +862,7 @@ describe("LocalMembership", () => {
const setScreenShareEnabled = vi.fn().mockRejectedValue(error); const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
const connection = { const connection = {
state$: constant(ConnectionState.LivekitConnected), state$: constant(ConnectionState.LivekitConnected),
transport: aTransport, transport: mockTransportConfig,
livekitRoom: mockLivekitRoom({ livekitRoom: mockLivekitRoom({
localParticipant: mockLocalParticipant({ localParticipant: mockLocalParticipant({
isScreenShareEnabled: false, isScreenShareEnabled: false,
src/state/CallViewModel/localMember/LocalMember.ts
+90 -117
View File
@@ -38,6 +38,10 @@ import {
startWith, startWith,
switchMap, switchMap,
tap, tap,
NEVER,
concat,
race,
Subject,
} from "rxjs"; } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { deepCompare } from "matrix-js-sdk/lib/utils"; import { deepCompare } from "matrix-js-sdk/lib/utils";
@@ -77,8 +81,6 @@ import {
} from "../remoteMembers/Connection.ts"; } from "../remoteMembers/Connection.ts";
import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { type HomeserverConnected } from "./HomeserverConnected.ts";
import { type LocalTransport } from "./LocalTransport.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"; import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
export enum TransportState { export enum TransportState {
@@ -148,7 +150,7 @@ interface Props {
homeserverConnected: HomeserverConnected; homeserverConnected: HomeserverConnected;
roomId: string; roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts; ownMembershipIdentity: CallMembershipIdentityParts;
localTransport: LocalTransport; localTransport$: Observable<LocalTransport>;
client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">; client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
matrixRTCSession: Pick< matrixRTCSession: Pick<
MatrixRTCSession, MatrixRTCSession,
@@ -194,7 +196,7 @@ interface Props {
export const createLocalMembership$ = ({ export const createLocalMembership$ = ({
scope, scope,
connectionManager, connectionManager,
localTransport, localTransport$: localTransportWithErrors$,
homeserverConnected, homeserverConnected,
createPublisherFactory, createPublisherFactory,
joinMatrixRTC, joinMatrixRTC,
@@ -252,20 +254,24 @@ export const createLocalMembership$ = ({
const logger = parentLogger.getChild("[LocalMembership]"); const logger = parentLogger.getChild("[LocalMembership]");
logger.debug(`Creating local membership..`); logger.debug(`Creating local membership..`);
// We consider error on the transport as fatal. // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
// Whether it is the active transport or the preferred transport. const fatalTransportError$ = new Subject<ElementCallError>();
const handleTransportError = (e: unknown): Observable<null> => { const localTransport$ = localTransportWithErrors$.pipe(
let error: ElementCallError; catchError((e: unknown) => {
if (e instanceof ElementCallError) { let error: ElementCallError;
error = e; if (e instanceof ElementCallError) {
} else { error = e;
error = new UnknownCallError( } else {
e instanceof Error ? e : new Error("Unknown error from localTransport"), error = new UnknownCallError(
); e instanceof Error
} ? e
setTransportError(error); : new Error("Unknown error from localTransport"),
return of(null); );
}; }
fatalTransportError$.next(error);
return NEVER; // Make this Observable swallow the error
}),
);
async function checkDelegationSupport( async function checkDelegationSupport(
endpointUrl: string, endpointUrl: string,
@@ -303,58 +309,40 @@ export const createLocalMembership$ = ({
// The transport that we will advertise in our membership, paired with info as // The transport that we will advertise in our membership, paired with info as
// to whether delayed event delegation is supported // to whether delayed event delegation is supported
const joinParams$ = localTransport.advertised$.pipe( const joinParams$ = scope.behavior(
catchError(handleTransportError), localTransport$.pipe(
distinctUntilChanged(areLivekitTransportsEqual), switchMap(async ({ transport }) => {
switchMap((transport) => { const transportSupportsDelegation = checkDelegationSupport(
if (transport === null) return of(null); transport.livekit_service_url + "/delegate_delayed_leave",
const transportSupportsDelegation = checkDelegationSupport( `transport ${transport.livekit_service_url}`,
transport.livekit_service_url + "/delegate_delayed_leave", );
`transport ${transport.livekit_service_url}`, return {
); transport,
return or$( delegationSupported:
from(homeserverSupportsDelegation), (await homeserverSupportsDelegation) ||
from(transportSupportsDelegation), (await transportSupportsDelegation),
).pipe( };
map((delegationSupported) => ({ transport, delegationSupported })),
startWith(null),
);
}),
);
// Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
const activeTransport$ = scope.behavior(
combineLatest([localTransport.active$, localTransport.advertised$]).pipe(
map(([active, advertised]) => {
// Our policy is to not publish to another transport if our prefered transport is miss-configured
if (advertised == null) return null;
return active?.transport ?? null;
}), }),
catchError(handleTransportError),
distinctUntilChanged(areLivekitTransportsEqual),
), ),
null,
); );
// Drop Epoch data here since we will not combine this anymore // Drop Epoch data here since we will not combine this anymore
const localConnection$ = scope.behavior( const localConnection$ = scope.behavior(
combineLatest([ combineLatest([
connectionManager.connectionManagerData$, connectionManager.connectionManagerData$,
activeTransport$, localTransport$,
]).pipe( ]).pipe(
map(([{ value: connectionData }, localTransport]) => { map(([{ value: connectionData }, { transport }]) =>
if (localTransport === null) { connectionData.getConnectionForTransport(transport),
return null; ),
}
return connectionData.getConnectionForTransport(localTransport);
}),
tap((connection) => { tap((connection) => {
logger.info( logger.info(
`Local connection updated: ${connection?.transport?.livekit_service_url}`, `Local connection updated: ${connection?.transport?.livekit_service_url}`,
); );
}), }),
), ),
null,
); );
// Tracks error that happen when creating the local tracks. // Tracks error that happen when creating the local tracks.
@@ -490,18 +478,6 @@ export const createLocalMembership$ = ({
} }
}; };
const fatalTransportError$ = new BehaviorSubject<ElementCallError | null>(
null,
);
const setTransportError = (e: ElementCallError): void => {
if (fatalTransportError$.value !== null) {
logger.error("Multiple Transport Errors:", e);
} else {
fatalTransportError$.next(e);
}
};
const localConnectionState$ = localConnection$.pipe( const localConnectionState$ = localConnection$.pipe(
switchMap((connection) => (connection ? connection.state$ : of(null))), switchMap((connection) => (connection ? connection.state$ : of(null))),
); );
@@ -509,38 +485,29 @@ export const createLocalMembership$ = ({
const mediaState$: Behavior<LocalMemberMediaState> = scope.behavior( const mediaState$: Behavior<LocalMemberMediaState> = scope.behavior(
combineLatest([ combineLatest([
localConnectionState$, localConnectionState$,
activeTransport$,
joinAndPublishRequested$, joinAndPublishRequested$,
from(trackStartRequested.promise).pipe( from(trackStartRequested.promise).pipe(
map(() => true), map(() => true),
startWith(false), startWith(false),
), ),
]).pipe( ]).pipe(
map( map(([localConnectionState, shouldPublish, shouldStartTracks]) => {
([ const trackState: TrackState = shouldStartTracks
localConnectionState, ? TrackState.Ready
localTransport, : TrackState.WaitingForUser;
shouldPublish,
shouldStartTracks,
]) => {
if (!localTransport) return null;
const trackState: TrackState = shouldStartTracks
? TrackState.Ready
: TrackState.WaitingForUser;
if ( if (
localConnectionState !== ConnectionState.LivekitConnected || localConnectionState !== ConnectionState.LivekitConnected ||
trackState !== TrackState.Ready trackState !== TrackState.Ready
) )
return { return {
connection: localConnectionState, connection: localConnectionState,
tracks: trackState, tracks: trackState,
}; };
if (!shouldPublish) return PublishState.WaitingForUser; if (!shouldPublish) return PublishState.WaitingForUser;
// if (!publishing) return PublishState.Starting; // if (!publishing) return PublishState.Starting;
return PublishState.Publishing; return PublishState.Publishing;
}, }),
),
distinctUntilChanged(deepCompare), distinctUntilChanged(deepCompare),
), ),
); );
@@ -554,30 +521,36 @@ export const createLocalMembership$ = ({
}; };
const localMemberState$ = scope.behavior<LocalMemberState>( const localMemberState$ = scope.behavior<LocalMemberState>(
combineLatest([ concat(
mediaState$, // Waiting until…
homeserverConnected.rtsSession$, of(TransportState.Waiting),
fatalMatrixError$, race(
fatalTransportError$, // either there is a fatal transport error
publishError$, fatalTransportError$,
]).pipe( // or the transport is available.
map( localTransport$.pipe(
([ switchMap(() =>
mediaState, // Once available, track session/media state.
rtcSessionStatus, combineLatest([
fatalMatrixError, mediaState$,
fatalTransportError, homeserverConnected.rtsSession$,
publishError, fatalMatrixError$,
]) => { publishError$,
if (fatalTransportError !== null) return fatalTransportError; ]).pipe(
// `mediaState` will be 'null' until the transport/connection appears. map(
if (mediaState && rtcSessionStatus) ([
return { mediaState,
matrix: fatalMatrixError ?? rtcSessionStatus, rtcSessionStatus,
media: publishError ?? mediaState, fatalMatrixError,
}; publishError,
return TransportState.Waiting; ]) => ({
}, matrix: fatalMatrixError ?? rtcSessionStatus,
media: publishError ?? mediaState,
}),
),
),
),
),
), ),
), ),
); );
@@ -685,7 +658,7 @@ export const createLocalMembership$ = ({
}); });
}); });
// Keep matrix rtc session in sync with advertisedTransport$, connectRequested$ // Join and leave the session as needed
scope.reconcile( scope.reconcile(
scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])), scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])),
async ([joinParams, shouldConnect]) => { async ([joinParams, shouldConnect]) => {
src/state/CallViewModel/localMember/LocalTransport.test.ts
+98 -169
View File
@@ -14,19 +14,10 @@ import {
type MockedObject, type MockedObject,
vi, vi,
} from "vitest"; } from "vitest";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { lastValueFrom } from "rxjs";
import fetchMock from "fetch-mock"; import fetchMock from "fetch-mock";
import { import { mockConfig, ownMemberMock } from "../../../utils/test";
mockConfig, import { getLocalTransport } from "./LocalTransport";
flushPromises,
ownMemberMock,
testScope,
} from "../../../utils/test";
import { createLocalTransport$ } from "./LocalTransport";
import { constant } from "../../Behavior";
import { Epoch, ObservableScope } from "../../ObservableScope";
import { import {
MatrixRTCTransportMissingError, MatrixRTCTransportMissingError,
FailToGetOpenIdToken, FailToGetOpenIdToken,
@@ -47,118 +38,79 @@ describe("LocalTransport", () => {
beforeEach(() => vi.clearAllMocks()); beforeEach(() => vi.clearAllMocks());
it("throws if config is missing", async () => { it("throws if config is missing", async () => {
const { advertised$, active$ } = createLocalTransport$({ await expect(
scope: testScope(), getLocalTransport({
roomId: "!room:example.org", roomId: "!room:example.org",
memberships$: constant(new Epoch<CallMembership[]>([])), client: {
client: { // eslint-disable-next-line @typescript-eslint/naming-convention
// eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: async () => Promise.resolve([]),
_unstable_getRTCTransports: async () => Promise.resolve([]), getDomain: () => "example.org",
getDomain: () => "example.org", // These won't be called in this error path but satisfy the type
// These won't be called in this error path but satisfy the type getOpenIdToken: vi.fn(),
getOpenIdToken: vi.fn(), getDeviceId: vi.fn(),
getDeviceId: vi.fn(), },
}, ownMembershipIdentity: ownMemberMock,
ownMembershipIdentity: ownMemberMock, matrixRTCMode: MatrixRTCMode.Compatibility,
matrixRTCMode: MatrixRTCMode.Compatibility, }),
}); ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
await flushPromises();
expect(() => advertised$.value).toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
expect(() => active$.value).toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
}); });
it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => { it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
// Provide a valid config so makeTransportInternal resolves a transport // Provide a valid config so makeTransportInternal resolves a transport
const scope = new ObservableScope();
mockConfig({ mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" }, livekit: { livekit_service_url: "https://lk.example.org" },
}); });
const resolver = Promise.withResolvers<void>(); vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockImplementation( new FailToGetOpenIdToken(new Error("no openid")),
async () => {
await resolver.promise;
throw new FailToGetOpenIdToken(new Error("no openid"));
},
); );
const observations: unknown[] = [];
const errors: Error[] = [];
const { advertised$, active$ } = createLocalTransport$({
scope,
roomId: "!example_room_id",
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: () => "example.org",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
});
active$.subscribe(
(o) => observations.push(o),
(e) => errors.push(e),
);
resolver.resolve();
await flushPromises();
const expectedError = new FailToGetOpenIdToken(new Error("no openid")); await expect(
expect(observations).toStrictEqual([null]); getLocalTransport({
expect(errors).toStrictEqual([expectedError]); roomId: "!example_room_id",
expect(() => advertised$.value).toThrow(expectedError); client: {
expect(() => active$.value).toThrow(expectedError); 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 // Use config so transport discovery succeeds, but delay OpenID JWT fetch
mockConfig({ mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" }, livekit: { livekit_service_url: "https://lk.example.org" },
}); });
const openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>(); vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
openIdResolver.promise,
);
const { advertised$, active$ } = createLocalTransport$({
scope: testScope(),
roomId: "!room:example.org",
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
});
openIdResolver.resolve?.({
url: "https://lk.example.org", url: "https://lk.example.org",
jwt: "jwt", jwt: "jwt",
livekitAlias: "Akph4alDMhen", livekitAlias: "Akph4alDMhen",
livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId, livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId,
}); });
expect(advertised$.value).toBe(null);
expect(active$.value).toBe(null); expect(
await flushPromises(); await getLocalTransport({
// final roomId: "!room:example.org",
const expectedTransport = { client: {
livekit_service_url: "https://lk.example.org", // eslint-disable-next-line @typescript-eslint/naming-convention
type: "livekit", _unstable_getRTCTransports: async () => Promise.resolve([]),
}; getDomain: () => "example.org",
expect(advertised$.value).toStrictEqual(expectedTransport); getOpenIdToken: vi.fn(),
expect(active$.value).toStrictEqual({ getDeviceId: vi.fn(),
transport: expectedTransport, },
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
}),
).toStrictEqual({
transport: {
livekit_service_url: "https://lk.example.org",
type: "livekit",
},
sfuConfig: { sfuConfig: {
jwt: "jwt", jwt: "jwt",
livekitAlias: "Akph4alDMhen", livekitAlias: "Akph4alDMhen",
@@ -168,22 +120,19 @@ describe("LocalTransport", () => {
}); });
}); });
type LocalTransportProps = Parameters<typeof createLocalTransport$>[0]; type LocalTransportProps = Parameters<typeof getLocalTransport>[0];
describe("transport configuration mechanisms", () => { describe("transport configuration mechanisms", () => {
let localTransportOpts: LocalTransportProps & { let localTransportOpts: LocalTransportProps & {
client: MockedObject<LocalTransportProps["client"]>; client: MockedObject<LocalTransportProps["client"]>;
}; };
let openIdResolver: PromiseWithResolvers<openIDSFU.SFUConfig>;
beforeEach(() => { beforeEach(() => {
mockConfig({}); mockConfig({});
customLivekitUrl.setValue(customLivekitUrl.defaultValue); customLivekitUrl.setValue(customLivekitUrl.defaultValue);
localTransportOpts = { localTransportOpts = {
ownMembershipIdentity: ownMemberMock, ownMembershipIdentity: ownMemberMock,
scope: testScope(),
roomId: "!example_room_id", roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility, matrixRTCMode: MatrixRTCMode.Compatibility,
memberships$: constant(new Epoch<CallMembership[]>([])),
client: { client: {
getDomain: vi.fn().mockReturnValue("example.org"), getDomain: vi.fn().mockReturnValue("example.org"),
// eslint-disable-next-line @typescript-eslint/naming-convention // eslint-disable-next-line @typescript-eslint/naming-convention
@@ -192,10 +141,6 @@ describe("LocalTransport", () => {
getDeviceId: vi.fn(), getDeviceId: vi.fn(),
}, },
}; };
openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
openIdResolver.promise,
);
}); });
afterEach(() => { afterEach(() => {
@@ -206,19 +151,15 @@ describe("LocalTransport", () => {
mockConfig({ mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" }, livekit: { livekit_service_url: "https://lk.example.org" },
}); });
const { advertised$, active$ } = vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
createLocalTransport$(localTransportOpts); openIdResponse,
openIdResolver.resolve?.(openIdResponse); );
expect(advertised$.value).toBe(null);
expect(active$.value).toBe(null); expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
await flushPromises(); transport: {
const expectedTransport = { livekit_service_url: "https://lk.example.org",
livekit_service_url: "https://lk.example.org", type: "livekit",
type: "livekit", },
};
expect(advertised$.value).toStrictEqual(expectedTransport);
expect(active$.value).toStrictEqual({
transport: expectedTransport,
sfuConfig: { sfuConfig: {
jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
livekitAlias: "Akph4alDMhen", livekitAlias: "Akph4alDMhen",
@@ -230,12 +171,11 @@ describe("LocalTransport", () => {
it("supports getting transport via user settings", async () => { it("supports getting transport via user settings", async () => {
customLivekitUrl.setValue("https://lk.example.org"); customLivekitUrl.setValue("https://lk.example.org");
const { advertised$, active$ } = vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
createLocalTransport$(localTransportOpts); openIdResponse,
openIdResolver.resolve?.(openIdResponse); );
expect(advertised$.value).toBe(null);
await flushPromises(); expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
expect(active$.value).toStrictEqual({
transport: { transport: {
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
type: "livekit", type: "livekit",
@@ -253,19 +193,15 @@ describe("LocalTransport", () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" }, { type: "livekit", livekit_service_url: "https://lk.example.org" },
]); ]);
const { advertised$, active$ } = vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
createLocalTransport$(localTransportOpts); openIdResponse,
openIdResolver.resolve?.(openIdResponse); );
expect(advertised$.value).toBe(null);
expect(active$.value).toBe(null); expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
await flushPromises(); transport: {
const expectedTransport = { livekit_service_url: "https://lk.example.org",
livekit_service_url: "https://lk.example.org", type: "livekit",
type: "livekit", },
};
expect(advertised$.value).toStrictEqual(expectedTransport);
expect(active$.value).toStrictEqual({
transport: expectedTransport,
sfuConfig: { sfuConfig: {
jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
livekitAlias: "Akph4alDMhen", livekitAlias: "Akph4alDMhen",
@@ -279,38 +215,31 @@ describe("LocalTransport", () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" }, { type: "livekit", livekit_service_url: "https://lk.example.org" },
]); ]);
openIdResolver.reject( vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
new FailToGetOpenIdToken(new Error("Test driven error")), new FailToGetOpenIdToken(new Error("Test driven error")),
); );
await expect(async () =>
lastValueFrom(createLocalTransport$(localTransportOpts).active$), await expect(getLocalTransport(localTransportOpts)).rejects.toThrow(
).rejects.toThrow(expect.any(FailToGetOpenIdToken)); expect.any(FailToGetOpenIdToken),
);
}); });
it("throws if no options are available", async () => { it("throws if no options are available", async () => {
const { advertised$, active$ } = createLocalTransport$({ await expect(
scope: testScope(), getLocalTransport({
ownMembershipIdentity: ownMemberMock, ownMembershipIdentity: ownMemberMock,
roomId: "!example_room_id", roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility, matrixRTCMode: MatrixRTCMode.Compatibility,
memberships$: constant(new Epoch<CallMembership[]>([])), client: {
client: { getDomain: () => "example.org",
getDomain: () => "example.org", // eslint-disable-next-line @typescript-eslint/naming-convention
// eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: async () => Promise.resolve([]),
_unstable_getRTCTransports: async () => Promise.resolve([]), // These won't be called in this error path but satisfy the type
// These won't be called in this error path but satisfy the type getOpenIdToken: vi.fn(),
getOpenIdToken: vi.fn(), getDeviceId: vi.fn(),
getDeviceId: vi.fn(), },
}, }),
}); ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
await flushPromises();
expect(() => advertised$.value).toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
expect(() => active$.value).toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
}); });
}); });
}); });
src/state/CallViewModel/localMember/LocalTransport.ts
+35 -97
View File
@@ -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. Please see LICENSE in the repository root for full details.
*/ */
import { import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
type CallMembership,
type LivekitTransportConfig,
} from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk"; 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 { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; 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 { Config } from "../../../config/Config.ts";
import { import {
FailToGetOpenIdToken, FailToGetOpenIdToken,
@@ -27,19 +21,12 @@ import {
type SFUConfig, type SFUConfig,
type OpenIDClientParts, type OpenIDClientParts,
} from "../../../livekit/openIDSFU.ts"; } from "../../../livekit/openIDSFU.ts";
import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
import { customLivekitUrl } from "../../../settings/settings.ts"; import { customLivekitUrl } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type MatrixRTCMode } from "../../../config/ConfigOptions.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 { interface Props {
scope: ObservableScope;
ownMembershipIdentity: CallMembershipIdentityParts; ownMembershipIdentity: CallMembershipIdentityParts;
memberships$: Behavior<Epoch<CallMembership[]>>;
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> & client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts; OpenIDClientParts;
// Used by the jwt service to create the livekit room and compute the livekit alias. // Used by the jwt service to create the livekit room and compute the livekit alias.
@@ -68,114 +55,65 @@ interface Props {
// //
// 2. // 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! // 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; transport: LivekitTransportConfig;
sfuConfig: SFUConfig; sfuConfig: SFUConfig;
} }
export function isLocalTransportWithSFUConfig( export function isLocalTransport(
obj: LivekitTransportConfig | LocalTransportWithSFUConfig, obj: LivekitTransportConfig | LocalTransport,
): obj is LocalTransportWithSFUConfig { ): obj is LocalTransport {
return "transport" in obj && "sfuConfig" in obj; return "transport" in obj && "sfuConfig" in obj;
} }
export interface LocalTransport {
/**
* The transport to be advertised in our MatrixRTC membership. `null` when not
* yet fetched/validated.
*/
advertised$: Behavior<LivekitTransportConfig | null>;
/**
* The transport to connect to and publish media on. `null` when not yet known
* or available.
*/
active$: Behavior<LocalTransportWithSFUConfig | null>;
}
/** /**
* Connects to the JWT service and determines the 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. * @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. * 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) * (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 * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
*/ */
export const createLocalTransport$ = ({ export async function getLocalTransport({
scope,
memberships$,
ownMembershipIdentity, ownMembershipIdentity,
client, client,
roomId, roomId,
matrixRTCMode, matrixRTCMode,
}: Props): LocalTransport => { }: Props): Promise<LocalTransport> {
const logger = rootLogger.getChild("[LocalTransport]"); const logger = rootLogger.getChild("[LocalTransport]");
const discovery = new RtcTransportAutoDiscovery({
const transportDiscovery = new RtcTransportAutoDiscovery({
client: client, client: client,
resolvedConfig: Config.get(), resolvedConfig: Config.get(),
logger: logger, logger: logger,
}); });
const customUrl = customLivekitUrl.value$.value;
// Get the preferred transport from the current deployment. // Respect the user's custom URL, if set
const discoveredTransport$ = from( const transport: LivekitTransportConfig | null = customUrl
transportDiscovery.discoverPreferredTransport(), ? { type: "livekit", livekit_service_url: customUrl }
); : await discovery.discoverPreferredTransport();
const preferredConfig$ = customLivekitUrl.value$.pipe( if (transport === null)
switchMap((customUrl) => { throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
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),
);
const preferredTransport$ = preferredConfig$.pipe( try {
switchMap(async (transport) => { return await doOpenIdAndJWTFromUrl(
try { transport,
return await doOpenIdAndJWTFromUrl( matrixRTCMode,
transport, ownMembershipIdentity,
matrixRTCMode, roomId,
ownMembershipIdentity, client,
roomId, logger,
client, );
logger, } catch (e) {
); logger.error(
} catch (e) { `Failed to authenticate to transport ${transport.livekit_service_url}`,
logger.error( e,
`Failed to authenticate to transport ${transport.livekit_service_url}`, );
e, throw mapAuthErrorToUserFriendlyError(e);
); }
throw mapAuthErrorToUserFriendlyError(e); }
}
}),
);
// Always publish on and advertise the preferred transport.
return {
advertised$: scope.behavior(
preferredTransport$.pipe(
map((t) => t.transport),
distinctUntilChanged(areLivekitTransportsEqual),
),
null,
),
active$: scope.behavior(preferredTransport$, null),
};
};
/** /**
* Utility to ensure the user can authenticate with the SFU. * Utility to ensure the user can authenticate with the SFU.
@@ -199,7 +137,7 @@ async function doOpenIdAndJWTFromUrl(
roomId: string, roomId: string,
client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts, client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts,
logger?: Logger, logger?: Logger,
): Promise<LocalTransportWithSFUConfig> { ): Promise<LocalTransport> {
const sfuConfig = await getSFUConfigWithOpenID( const sfuConfig = await getSFUConfigWithOpenID(
client, client,
membership, membership,
src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts
+6 -6
View File
@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
*/ */
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; 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 LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
import { type RemoteParticipant } from "livekit-client"; import { type RemoteParticipant } from "livekit-client";
import { logger } from "matrix-js-sdk/lib/logger"; 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 { type Connection } from "./Connection.ts";
import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts"; import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { constant, type Behavior } from "../../Behavior.ts"; import { type Behavior } from "../../Behavior.ts";
// Some test constants // Some test constants
@@ -74,7 +74,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({ const { connectionManagerData$ } = createConnectionManager$({
scope: testScope, scope: testScope,
connectionFactory: fakeConnectionFactory, connectionFactory: fakeConnectionFactory,
localTransport$: constant(null), localTransport$: NEVER,
remoteTransports$: behavior("a", { remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0), a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}), }),
@@ -115,7 +115,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({ const { connectionManagerData$ } = createConnectionManager$({
scope: testScope, scope: testScope,
connectionFactory: fakeConnectionFactory, connectionFactory: fakeConnectionFactory,
localTransport$: constant(null), localTransport$: NEVER,
remoteTransports$: behavior("abcdef", { remoteTransports$: behavior("abcdef", {
a: new Epoch([TRANSPORT_1], 0), a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1], 1), b: new Epoch([TRANSPORT_1], 1),
@@ -162,7 +162,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({ const { connectionManagerData$ } = createConnectionManager$({
scope: testScope, scope: testScope,
connectionFactory: fakeConnectionFactory, connectionFactory: fakeConnectionFactory,
localTransport$: constant(null), localTransport$: NEVER,
remoteTransports$: behavior("abc", { remoteTransports$: behavior("abc", {
a: new Epoch([TRANSPORT_1], 0), a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1), b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
@@ -297,7 +297,7 @@ describe("connectionManagerData$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({ const { connectionManagerData$ } = createConnectionManager$({
scope: testScope, scope: testScope,
connectionFactory: fakeConnectionFactory, connectionFactory: fakeConnectionFactory,
localTransport$: constant(null), localTransport$: NEVER,
remoteTransports$: behavior("a", { remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0), a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}), }),
src/state/CallViewModel/remoteMembers/ConnectionManager.ts
+27 -13
View File
@@ -7,7 +7,16 @@ Please see LICENSE in the repository root for full details.
*/ */
import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc"; import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, map, of, switchMap } from "rxjs"; import {
combineLatest,
map,
type Observable,
of,
switchMap,
startWith,
catchError,
NEVER,
} from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { type RemoteParticipant } from "livekit-client"; import { type RemoteParticipant } from "livekit-client";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
@@ -19,8 +28,8 @@ import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { import {
isLocalTransportWithSFUConfig, isLocalTransport,
type LocalTransportWithSFUConfig, type LocalTransport,
} from "../localMember/LocalTransport.ts"; } from "../localMember/LocalTransport.ts";
import { type SFUConfig } from "../../../livekit/openIDSFU.ts"; import { type SFUConfig } from "../../../livekit/openIDSFU.ts";
@@ -78,7 +87,7 @@ export class ConnectionManagerData {
interface Props { interface Props {
scope: ObservableScope; scope: ObservableScope;
connectionFactory: ConnectionFactory; connectionFactory: ConnectionFactory;
localTransport$: Behavior<LocalTransportWithSFUConfig | null>; localTransport$: Observable<LocalTransport>;
remoteTransports$: Behavior<Epoch<LivekitTransportConfig[]>>; remoteTransports$: Behavior<Epoch<LivekitTransportConfig[]>>;
logger: Logger; logger: Logger;
@@ -121,6 +130,14 @@ export function createConnectionManager$({
const logger = parentLogger.getChild("[ConnectionManager]"); const logger = parentLogger.getChild("[ConnectionManager]");
// TODO logger: only construct one logger from the client and make it compatible via a EC specific sing // 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. * All transports currently managed by the ConnectionManager.
* *
@@ -130,16 +147,12 @@ export function createConnectionManager$({
* externally this is modified via `registerTransports()`. * externally this is modified via `registerTransports()`.
*/ */
const localAndRemoteTransports$: Behavior< const localAndRemoteTransports$: Behavior<
Epoch<(LivekitTransportConfig | LocalTransportWithSFUConfig)[]> Epoch<(LivekitTransportConfig | LocalTransport)[]>
> = scope.behavior( > = scope.behavior(
combineLatest([remoteTransports$, localTransport$]).pipe( combineLatest([localTransportAsArray$, remoteTransports$]).pipe(
// Combine local and remote transports into one transport array // Combine local and remote transports into one transport array
// and set the forceOldJwtEndpoint property on the local transport // and set the forceOldJwtEndpoint property on the local transport
map(([remoteTransports, localTransport]) => { map(([localTransportAsArray, remoteTransports]) => {
let localTransportAsArray: LocalTransportWithSFUConfig[] = [];
if (localTransport) {
localTransportAsArray = [localTransport];
}
const dedupedRemote = removeDuplicateTransports(remoteTransports.value); const dedupedRemote = removeDuplicateTransports(remoteTransports.value);
const remoteWithoutLocal = dedupedRemote.filter( const remoteWithoutLocal = dedupedRemote.filter(
(transport) => (transport) =>
@@ -170,8 +183,9 @@ export function createConnectionManager$({
"ConnectionManager connections$", "ConnectionManager connections$",
function* (transports) { function* (transports) {
for (const transport of transports) { for (const transport of transports) {
if (isLocalTransportWithSFUConfig(transport)) { if (isLocalTransport(transport)) {
// This is the local transport; only the `LocalTransportWithSFUConfig` has a `sfuConfig` field. // This is the local transport; only the `LocalTransport`
// interface has a `sfuConfig` field.
yield { yield {
keys: [ keys: [
transport.transport.livekit_service_url, transport.transport.livekit_service_url,
src/state/CallViewModel/remoteMembers/integration.test.ts
+2 -3
View File
@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
*/ */
import { test, vi, expect, beforeEach, afterEach } from "vitest"; 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 { type Room as LivekitRoom } from "livekit-client";
import EventEmitter from "events"; import EventEmitter from "events";
import fetchMock from "fetch-mock"; import fetchMock from "fetch-mock";
@@ -34,7 +34,6 @@ import {
} from "./MatrixLivekitMembers.ts"; } from "./MatrixLivekitMembers.ts";
import { createConnectionManager$ } from "./ConnectionManager.ts"; import { createConnectionManager$ } from "./ConnectionManager.ts";
import { membershipsAndTransports$ } from "../../SessionBehaviors.ts"; import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
import { constant } from "../../Behavior.ts";
import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts"; import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
// Test the integration of ConnectionManager and MatrixLivekitMerger // Test the integration of ConnectionManager and MatrixLivekitMerger
@@ -124,7 +123,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
const connectionManager = createConnectionManager$({ const connectionManager = createConnectionManager$({
scope: testScope, scope: testScope,
connectionFactory: ecConnectionFactory, connectionFactory: ecConnectionFactory,
localTransport$: constant(null), localTransport$: NEVER,
remoteTransports$: membershipsAndTransports.transports$, remoteTransports$: membershipsAndTransports.transports$,
logger: logger, logger: logger,
ownMembershipIdentity: ownMemberMock, ownMembershipIdentity: ownMemberMock,
src/state/SessionBehaviors.ts
+4 -8
View File
@@ -22,6 +22,10 @@ import {
} from "./ObservableScope"; } from "./ObservableScope";
import { type Behavior } from "./Behavior"; import { type Behavior } from "./Behavior";
/**
* Tracks the transports used by ourselves, plus all other MatrixRTC session
* members.
*/
export const membershipsAndTransports$ = ( export const membershipsAndTransports$ = (
scope: ObservableScope, scope: ObservableScope,
memberships$: Behavior<Epoch<CallMembership[]>>, memberships$: Behavior<Epoch<CallMembership[]>>,
@@ -31,14 +35,6 @@ export const membershipsAndTransports$ = (
>; >;
transports$: Behavior<Epoch<LivekitTransportConfig[]>>; transports$: Behavior<Epoch<LivekitTransportConfig[]>>;
} => { } => {
/**
* Lists the transports used by ourselves, plus all other MatrixRTC session
* members.
* For completeness this also lists the preferred transport and
* whether we are in multi-SFU mode or sticky events mode.
* `advertisedTransport$` reads these values together, so bundling them avoids inconsistent state or
* excessive updates when using RxJS.
*/
const membershipsWithTransport$: Behavior< const membershipsWithTransport$: Behavior<
Epoch< Epoch<
{ {
src/utils/observable.test.ts
+1 -22
View File
@@ -9,30 +9,9 @@ import { expect, test } from "vitest";
import { type Observable, of, Subject, switchMap } from "rxjs"; import { type Observable, of, Subject, switchMap } from "rxjs";
import { withTestScheduler } from "./test"; import { withTestScheduler } from "./test";
import { or$, filterBehavior, generateItems, pauseWhen } from "./observable"; import { filterBehavior, generateItems, pauseWhen } from "./observable";
import { type Behavior } from "../state/Behavior"; 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", () => { test("pauseWhen", () => {
withTestScheduler(({ behavior, expectObservable }) => { withTestScheduler(({ behavior, expectObservable }) => {
const inputMarbles = " abcdefgh-i-jk-"; const inputMarbles = " abcdefgh-i-jk-";
src/utils/observable.ts
-9
View File
@@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details.
import { import {
type Observable, type Observable,
audit, audit,
combineLatest,
concat, concat,
defer, defer,
filter, filter,
@@ -113,14 +112,6 @@ export function getValue<T>(state$: Observable<T>): T {
return value; return value;
} }
/**
* Creates an Observable that has a value of true whenever some of its inputs
* are true.
*/
export function or$(...inputs: Observable<boolean>[]): Observable<boolean> {
return combineLatest(inputs, (...flags) => flags.some((flag) => flag));
}
/** /**
* RxJS operator that pauses all changes in the input value whenever a Behavior * 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 * is true. When the Behavior returns to being false, the most recently