Make use of the new jwt service endpoint (with delayed event delegation)

This also does all the compatibility work. When to use which endpoint to
authenticate agains a jwt service.
This commit is contained in:
Timo K
2025-12-17 09:53:49 +01:00
parent 9bd51fdfc4
commit ab7e3486b3
17 changed files with 294 additions and 74 deletions

View File

@@ -7,9 +7,11 @@ Please see LICENSE in the repository root for full details.
import { type IOpenIDToken, type MatrixClient } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { FailToGetOpenIdToken } from "../utils/errors";
import { doNetworkOperationWithRetry } from "../utils/matrix";
import { Config } from "../config/Config";
export interface SFUConfig {
url: string;
@@ -33,8 +35,12 @@ export type OpenIDClientParts = Pick<
*/
export async function getSFUConfigWithOpenID(
client: OpenIDClientParts,
membership: CallMembershipIdentityParts,
serviceUrl: string,
matrixRoomId: string,
livekitRoomAlias: string,
matrix2jwt: boolean,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<SFUConfig> {
let openIdToken: IOpenIDToken;
try {
@@ -49,21 +55,31 @@ export async function getSFUConfigWithOpenID(
logger.debug("Got openID token", openIdToken);
logger.info(`Trying to get JWT for focus ${serviceUrl}...`);
const sfuConfig = await getLiveKitJWT(
client,
const args: [CallMembershipIdentityParts, string, string, IOpenIDToken] = [
membership,
serviceUrl,
matrixRoomId,
livekitRoomAlias,
openIdToken,
);
logger.info(`Got JWT from call's active focus URL.`);
return sfuConfig;
];
if (matrix2jwt) {
const sfuConfig = await getLiveKitJWTWithDelayDelegation(
...args,
delayEndpointBaseUrl,
delayId,
);
logger.info(`Got JWT from call's active focus URL.`);
return sfuConfig;
} else {
const sfuConfig = await getLiveKitJWT(...args);
logger.info(`Got JWT from call's active focus URL.`);
return sfuConfig;
}
}
async function getLiveKitJWT(
client: OpenIDClientParts,
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
roomName: string,
livekitRoomAlias: string,
openIDToken: IOpenIDToken,
): Promise<SFUConfig> {
try {
@@ -73,9 +89,9 @@ async function getLiveKitJWT(
"Content-Type": "application/json",
},
body: JSON.stringify({
room: roomName,
room: livekitRoomAlias,
openid_token: openIDToken,
device_id: client.getDeviceId(),
device_id: membership.deviceId,
}),
});
if (!res.ok) {
@@ -86,3 +102,53 @@ async function getLiveKitJWT(
throw new Error("SFU Config fetch failed with exception " + e);
}
}
export async function getLiveKitJWTWithDelayDelegation(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
livekitRoomAlias: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<SFUConfig> {
const { userId, deviceId, memberId } = membership;
const body = {
room_id: livekitRoomAlias,
slot_id: "m.call#ROOM",
openid_token: openIDToken,
member: {
id: memberId,
claimed_user_id: userId,
claimed_device_id: deviceId,
},
};
let bodyDalayParts = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
const delayTimeoutMs =
Config.get().matrix_rtc_session?.delayed_leave_event_delay_ms ?? 1000;
bodyDalayParts = {
delay_id: delayId,
delay_timeout: delayTimeoutMs,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
try {
const res = await fetch(livekitServiceURL + "/get_token", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ ...body, ...bodyDalayParts }),
});
if (!res.ok) {
throw new Error("SFU Config fetch failed with status code " + res.status);
}
return await res.json();
} catch (e) {
throw new Error("SFU Config fetch failed with exception " + e);
}
}

View File

@@ -41,10 +41,12 @@ import {
} from "rxjs";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import {
MembershipManagerEvent,
type LivekitTransport,
type MatrixRTCSession,
} from "matrix-js-sdk/lib/matrixrtc";
import { type IWidgetApiRequest } from "matrix-widget-api";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import {
LocalUserMediaViewModel,
@@ -98,7 +100,7 @@ import {
type SpotlightLandscapeLayoutMedia,
type SpotlightPortraitLayoutMedia,
} from "../layout-types.ts";
import { ElementCallError } from "../../utils/errors.ts";
import { ElementCallError, UnknownCallError } from "../../utils/errors.ts";
import { type ObservableScope } from "../ObservableScope.ts";
import { createHomeserverConnected$ } from "./localMember/HomeserverConnected.ts";
import {
@@ -375,8 +377,11 @@ export function createCallViewModel$(
trackProcessorState$: Behavior<ProcessorState>,
): CallViewModel {
const client = matrixRoom.client;
const userId = client.getUserId()!;
const deviceId = client.getDeviceId()!;
const userId = client.getUserId();
const deviceId = client.getDeviceId();
if (!(userId && deviceId))
throw new UnknownCallError(new Error("userId and deviceId are required"));
const livekitKeyProvider = getE2eeKeyProvider(
options.encryptionSystem,
matrixRTCSession,
@@ -407,10 +412,29 @@ export function createCallViewModel$(
memberships$,
);
const ownMembershipIdentity: CallMembershipIdentityParts = {
userId,
deviceId,
memberId: `${userId}:${deviceId}`,
};
const localTransport$ = createLocalTransport$({
scope: scope,
memberships$: memberships$,
ownMembershipIdentity,
client,
useMatrix2$: scope.behavior(
options.matrixRTCMode$.pipe(map((v) => v === MatrixRTCMode.Matrix_2_0)),
),
delayId$: scope.behavior(
(
fromEvent(
matrixRTCSession,
MembershipManagerEvent.DelayIdChanged,
) as Observable<string | undefined>
).pipe(map((v) => v ?? null)),
matrixRTCSession.delayId ?? null,
),
roomId: matrixRoom.roomId,
useOldestMember$: scope.behavior(
options.matrixRTCMode$.pipe(map((v) => v === MatrixRTCMode.Legacy)),
@@ -455,6 +479,7 @@ export function createCallViewModel$(
),
),
logger: logger,
ownMembershipIdentity,
});
const matrixLivekitMembers$ = createMatrixLivekitMembers$({
@@ -485,6 +510,7 @@ export function createCallViewModel$(
joinMatrixRTC: (transport: LivekitTransport) => {
return enterRTCSession(
matrixRTCSession,
ownMembershipIdentity,
transport,
connectOptions$.value,
);

View File

@@ -24,6 +24,7 @@ import {
mockLivekitRoom,
mockMuteStates,
withTestScheduler,
ownMemberMock,
} from "../../../utils/test";
import {
TransportState,
@@ -108,6 +109,7 @@ describe("LocalMembership", () => {
enterRTCSession(
mockedSession,
ownMemberMock,
{
livekit_alias: "roomId",
livekit_service_url: "http://my-well-known-service-url.com",
@@ -166,6 +168,7 @@ describe("LocalMembership", () => {
enterRTCSession(
mockedSession,
ownMemberMock,
{
livekit_alias: "roomId",
livekit_service_url: "http://my-well-known-service-url.com",

View File

@@ -34,6 +34,7 @@ import {
} from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { deepCompare } from "matrix-js-sdk/lib/utils";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { constant, type Behavior } from "../../Behavior.ts";
import { type IConnectionManager } from "../remoteMembers/ConnectionManager.ts";
@@ -657,6 +658,7 @@ interface EnterRTCSessionOptions {
// Exported for unit testing
export function enterRTCSession(
rtcSession: MatrixRTCSession,
ownMembershipIdentity: CallMembershipIdentityParts,
transport: LivekitTransport,
{ encryptMedia, matrixRTCMode }: EnterRTCSessionOptions,
): void {
@@ -674,7 +676,8 @@ export function enterRTCSession(
const multiSFU = matrixRTCMode !== MatrixRTCMode.Legacy;
// Multi-sfu does not need a preferred foci list. just the focus that is actually used.
// TODO where/how do we track errors originating from the ongoing rtcSession?
rtcSession.joinRoomSession(
rtcSession.joinRTCSession(
ownMembershipIdentity,
multiSFU ? [] : [transport],
multiSFU ? transport : undefined,
{

View File

@@ -9,7 +9,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject } from "rxjs";
import { mockConfig, flushPromises } from "../../../utils/test";
import { mockConfig, flushPromises, ownMemberMock } from "../../../utils/test";
import { createLocalTransport$ } from "./LocalTransport";
import { constant } from "../../Behavior";
import { Epoch, ObservableScope } from "../../ObservableScope";
@@ -32,10 +32,14 @@ describe("LocalTransport", () => {
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: () => "",
baseUrl: "example.org",
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
useMatrix2$: constant(false),
delayId$: constant("delay_id_mock"),
});
await flushPromises();
@@ -65,11 +69,15 @@ describe("LocalTransport", () => {
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
baseUrl: "https://lk.example.org",
// Use empty domain to skip .well-known and use config directly
getDomain: () => "",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
useMatrix2$: constant(false),
delayId$: constant("delay_id_mock"),
});
localTransport$.subscribe(
(o) => observations.push(o),
@@ -105,7 +113,11 @@ describe("LocalTransport", () => {
getDomain: () => "",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
baseUrl: "https://lk.example.org",
},
ownMembershipIdentity: ownMemberMock,
useMatrix2$: constant(false),
delayId$: constant("delay_id_mock"),
});
openIdResolver.resolve?.({ url: "https://lk.example.org", jwt: "jwt" });
@@ -140,7 +152,11 @@ describe("LocalTransport", () => {
getDomain: () => "",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
baseUrl: "https://lk.example.org",
},
ownMembershipIdentity: ownMemberMock,
useMatrix2$: constant(false),
delayId$: constant("delay_id_mock"),
});
openIdResolver.resolve?.({ url: "https://lk.example.org", jwt: "jwt" });

View File

@@ -23,6 +23,7 @@ import {
} from "rxjs";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Behavior } from "../../Behavior.ts";
import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
@@ -34,6 +35,7 @@ import {
} from "../../../livekit/openIDSFU.ts";
import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
import { customLivekitUrl } from "../../../settings/settings.ts";
import { type LivekitTransportWithVersion } from "../remoteMembers/ConnectionManager.ts";
const logger = rootLogger.getChild("[LocalTransport]");
@@ -44,10 +46,13 @@ const logger = rootLogger.getChild("[LocalTransport]");
*/
interface Props {
scope: ObservableScope;
ownMembershipIdentity: CallMembershipIdentityParts;
memberships$: Behavior<Epoch<CallMembership[]>>;
client: Pick<MatrixClient, "getDomain"> & OpenIDClientParts;
client: Pick<MatrixClient, "getDomain" | "baseUrl"> & OpenIDClientParts;
roomId: string;
useOldestMember$: Behavior<boolean>;
useMatrix2$: Behavior<boolean>;
delayId$: Behavior<string | null>;
}
/**
@@ -62,20 +67,26 @@ interface Props {
export const createLocalTransport$ = ({
scope,
memberships$,
ownMembershipIdentity,
client,
roomId,
useOldestMember$,
}: Props): Behavior<LivekitTransport | null> => {
useMatrix2$,
delayId$,
}: Props): Behavior<LivekitTransportWithVersion | null> => {
/**
* The transport over which we should be actively publishing our media.
* undefined when not joined.
*/
const oldestMemberTransport$ = scope.behavior(
memberships$.pipe(
map(
(memberships) =>
memberships.value[0]?.getTransport(memberships.value[0]) ?? null,
),
map((memberships) => {
const oldestMember = memberships.value[0];
const t = oldestMember?.getTransport(memberships.value[0]);
if (!t) return null;
// Here we will use the matrix2 information from the oldest member transport.
return { ...t, useMatrix2: oldestMember.kind === "rtc" };
}),
first((t) => t != null && isLivekitTransport(t)),
),
null,
@@ -87,12 +98,24 @@ export const createLocalTransport$ = ({
*
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
*/
const preferredTransport$: Behavior<LivekitTransport | null> = scope.behavior(
customLivekitUrl.value$.pipe(
switchMap((customUrl) => from(makeTransport(client, roomId, customUrl))),
),
null,
);
const preferredTransport$: Behavior<LivekitTransportWithVersion | null> =
scope.behavior(
combineLatest([customLivekitUrl.value$, useMatrix2$, delayId$]).pipe(
switchMap(([customUrl, useMatrix2, delayId]) =>
from(
makeTransport(
client,
ownMembershipIdentity,
roomId,
customUrl,
useMatrix2,
delayId ?? undefined,
),
),
),
),
null,
);
/**
* The chosen transport we should advertise in our MatrixRTC membership.
@@ -123,10 +146,13 @@ const FOCI_WK_KEY = "org.matrix.msc4143.rtc_foci";
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
*/
async function makeTransport(
client: Pick<MatrixClient, "getDomain"> & OpenIDClientParts,
client: Pick<MatrixClient, "getDomain" | "baseUrl"> & OpenIDClientParts,
membership: CallMembershipIdentityParts,
roomId: string,
urlFromDevSettings: string | null,
): Promise<LivekitTransport> {
matrix2jwt = false,
delayId?: string,
): Promise<LivekitTransportWithVersion> {
let transport: LivekitTransport | undefined;
logger.trace("Searching for a preferred transport");
//TODO refactor this to use the jwt service returned alias.
@@ -176,13 +202,18 @@ async function makeTransport(
transport = transportFromConf;
}
if (!transport) throw new MatrixRTCTransportMissingError(domain ?? ""); // this will call the jwt/sfu/get endpoint to pre create the livekit room.
if (!transport) throw new MatrixRTCTransportMissingError(domain ?? "");
// this will call the jwt/sfu/get endpoint to pre create the livekit room.
await getSFUConfigWithOpenID(
client,
membership,
transport.livekit_service_url,
transport.livekit_alias,
matrix2jwt,
client.baseUrl,
delayId,
);
return transport;
return { ...transport, useMatrix2: matrix2jwt };
}

View File

@@ -27,7 +27,6 @@ import EventEmitter from "events";
import { type IOpenIDToken } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import type { LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
Connection,
ConnectionState,
@@ -39,7 +38,8 @@ import {
ElementCallError,
FailToGetOpenIdToken,
} from "../../../utils/errors.ts";
import { mockRemoteParticipant } from "../../../utils/test.ts";
import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts";
import { type LivekitTransportWithVersion } from "./ConnectionManager.ts";
let testScope: ObservableScope;
@@ -50,10 +50,11 @@ let fakeLivekitRoom: MockedObject<LivekitRoom>;
let localParticipantEventEmiter: EventEmitter;
let fakeLocalParticipant: MockedObject<LocalParticipant>;
const livekitFocus: LivekitTransport = {
const livekitFocus: LivekitTransportWithVersion = {
livekit_alias: "!roomID:example.org",
livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
type: "livekit",
useMatrix2: false,
};
function setupTest(): void {
@@ -137,7 +138,7 @@ function setupRemoteConnection(): Connection {
return Promise.resolve();
});
return new Connection(opts, logger);
return new Connection(opts, logger, ownMemberMock);
}
afterEach(() => {
@@ -156,7 +157,7 @@ describe("Start connection states", () => {
scope: testScope,
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const connection = new Connection(opts, logger, ownMemberMock);
expect(connection.state$.getValue()).toEqual("Initialized");
});
@@ -172,7 +173,7 @@ describe("Start connection states", () => {
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const connection = new Connection(opts, logger, ownMemberMock);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
@@ -222,7 +223,7 @@ describe("Start connection states", () => {
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const connection = new Connection(opts, logger, ownMemberMock);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
@@ -279,7 +280,7 @@ describe("Start connection states", () => {
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const connection = new Connection(opts, logger, ownMemberMock);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {

View File

@@ -18,6 +18,7 @@ import {
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, map } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import {
getSFUConfigWithOpenID,
@@ -35,7 +36,7 @@ import {
export interface ConnectionOpts {
/** The media transport to connect to. */
transport: LivekitTransport;
transport: LivekitTransport & { useMatrix2: boolean };
/** The Matrix client to use for OpenID and SFU config requests. */
client: OpenIDClientParts;
/** The observable scope to use for this connection. */
@@ -88,7 +89,7 @@ export class Connection {
/**
* The media transport to connect to.
*/
public readonly transport: LivekitTransport;
public readonly transport: LivekitTransport & { useMatrix2: boolean };
public readonly livekitRoom: LivekitRoom;
@@ -189,9 +190,18 @@ export class Connection {
protected async getSFUConfigWithOpenID(): Promise<SFUConfig> {
return await getSFUConfigWithOpenID(
this.client,
this.ownMembershipIdentity,
this.transport.livekit_service_url,
this.transport.livekit_alias,
this.transport.useMatrix2,
);
// client: OpenIDClientParts,
// membership: CallMembershipIdentityParts,
// serviceUrl: string,
// livekitRoomAlias: string,
// matrix2jwt: boolean,
// delayEndpointBaseUrl?: string,
// delayId?: string,
}
/**
@@ -220,7 +230,11 @@ export class Connection {
*
* @param logger
*/
public constructor(opts: ConnectionOpts, logger: Logger) {
public constructor(
opts: ConnectionOpts,
logger: Logger,
private ownMembershipIdentity: CallMembershipIdentityParts,
) {
this.logger = logger.getChild("[Connection]");
this.logger.info(
`[Connection] Creating new connection to ${opts.transport.livekit_service_url} ${opts.transport.livekit_alias}`,

View File

@@ -5,7 +5,6 @@ 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 {
Room as LivekitRoom,
type RoomOptions,
@@ -15,6 +14,7 @@ import {
} from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import E2EEWorker from "livekit-client/e2ee-worker?worker";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type ObservableScope } from "../../ObservableScope.ts";
import { Connection } from "./Connection.ts";
@@ -23,13 +23,15 @@ import type { MediaDevices } from "../../MediaDevices.ts";
import type { Behavior } from "../../Behavior.ts";
import type { ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
import { defaultLiveKitOptions } from "../../../livekit/options.ts";
import { type LivekitTransportWithVersion } from "./ConnectionManager.ts";
// TODO evaluate if this should be done like the Publisher Factory
export interface ConnectionFactory {
createConnection(
transport: LivekitTransport,
transport: LivekitTransportWithVersion,
scope: ObservableScope,
logger: Logger,
ownMembershipIdentity: CallMembershipIdentityParts,
): Connection;
}
@@ -77,10 +79,19 @@ export class ECConnectionFactory implements ConnectionFactory {
this.livekitRoomFactory = livekitRoomFactory ?? defaultFactory;
}
/**
*
* @param transport
* @param scope
* @param logger
* @param ownMembershipIdentity required to connect (using the jwt service) with the SFU.
* @returns
*/
public createConnection(
transport: LivekitTransport,
transport: LivekitTransportWithVersion,
scope: ObservableScope,
logger: Logger,
ownMembershipIdentity: CallMembershipIdentityParts,
): Connection {
return new Connection(
{
@@ -90,6 +101,7 @@ export class ECConnectionFactory implements ConnectionFactory {
livekitRoomFactory: this.livekitRoomFactory,
},
logger,
ownMembershipIdentity,
);
}
}

View File

@@ -14,26 +14,29 @@ import { logger } from "matrix-js-sdk/lib/logger";
import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
import {
createConnectionManager$,
type LivekitTransportWithVersion,
type ConnectionManagerData,
} from "./ConnectionManager.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
import { withTestScheduler } from "../../../utils/test.ts";
import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type Behavior } from "../../Behavior.ts";
// Some test constants
const TRANSPORT_1: LivekitTransport = {
const TRANSPORT_1: LivekitTransportWithVersion = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
livekit_alias: "!alias:example.org",
useMatrix2: false,
};
const TRANSPORT_2: LivekitTransport = {
const TRANSPORT_2: LivekitTransportWithVersion = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
livekit_alias: "!alias:sample.com",
useMatrix2: false,
};
let fakeConnectionFactory: ConnectionFactory;
@@ -80,6 +83,7 @@ describe("connections$ stream", () => {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
@@ -124,6 +128,7 @@ describe("connections$ stream", () => {
f: new Epoch([TRANSPORT_1, TRANSPORT_2], 5),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
@@ -166,6 +171,7 @@ describe("connections$ stream", () => {
c: new Epoch([TRANSPORT_1], 2),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
@@ -279,6 +285,7 @@ describe("connectionManagerData$ stream", () => {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(connectionManagerData$).toBe("abcd", {

View File

@@ -10,6 +10,7 @@ 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";
@@ -18,6 +19,10 @@ import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
export type LivekitTransportWithVersion = LivekitTransport & {
useMatrix2: boolean;
};
export class ConnectionManagerData {
private readonly store: Map<string, [Connection, RemoteParticipant[]]> =
new Map();
@@ -59,8 +64,9 @@ export class ConnectionManagerData {
interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
inputTransports$: Behavior<Epoch<LivekitTransport[]>>;
inputTransports$: Behavior<Epoch<LivekitTransportWithVersion[]>>;
logger: Logger;
ownMembershipIdentity: CallMembershipIdentityParts;
}
// TODO - write test for scopes (do we really need to bind scope)
@@ -87,6 +93,7 @@ export function createConnectionManager$({
connectionFactory,
inputTransports$,
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
@@ -119,20 +126,26 @@ export function createConnectionManager$({
function* (transports) {
for (const transport of transports)
yield {
keys: [transport.livekit_service_url, transport.livekit_alias],
keys: [
transport.livekit_service_url,
transport.livekit_alias,
transport.useMatrix2,
],
data: undefined,
};
},
(scope, _data$, serviceUrl, alias) => {
(scope, _data$, serviceUrl, alias, useMatrix2) => {
logger.debug(`Creating connection to ${serviceUrl} (${alias})`);
const connection = connectionFactory.createConnection(
{
type: "livekit",
livekit_service_url: serviceUrl,
livekit_alias: alias,
useMatrix2,
},
scope,
logger,
ownMembershipIdentity,
);
// Start the connection immediately
// Use connection state to track connection progress
@@ -187,12 +200,12 @@ export function createConnectionManager$({
return { connectionManagerData$ };
}
function removeDuplicateTransports(
transports: LivekitTransport[],
): LivekitTransport[] {
function removeDuplicateTransports<T extends LivekitTransport>(
transports: T[],
): T[] {
return transports.reduce((acc, transport) => {
if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
acc.push(transport);
return acc;
}, [] as LivekitTransport[]);
}, [] as T[]);
}

View File

@@ -15,7 +15,11 @@ import EventEmitter from "events";
import { ObservableScope } from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import type { OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import { exampleTransport, mockMediaDevices } from "../../../utils/test.ts";
import {
exampleTransport,
mockMediaDevices,
ownMemberMock,
} from "../../../utils/test.ts";
import type { ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
import { constant } from "../../Behavior";
@@ -72,7 +76,12 @@ describe("ECConnectionFactory - Audio inputs options", () => {
echo,
noise,
);
ecConnectionFactory.createConnection(exampleTransport, testScope, logger);
ecConnectionFactory.createConnection(
exampleTransport,
testScope,
logger,
ownMemberMock,
);
// Check if Room was constructed with expected options
expect(RoomConstructor).toHaveBeenCalledWith(
@@ -113,7 +122,12 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
false,
false,
);
ecConnectionFactory.createConnection(exampleTransport, testScope, logger);
ecConnectionFactory.createConnection(
exampleTransport,
testScope,
logger,
ownMemberMock,
);
// Check if Room was constructed with expected options
expect(RoomConstructor).toHaveBeenCalledWith(

View File

@@ -176,9 +176,9 @@ export function createMatrixLivekitMembers$({
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
// TODO add this to the JS-SDK
export function areLivekitTransportsEqual(
t1: LivekitTransport | null,
t2: LivekitTransport | null,
export function areLivekitTransportsEqual<T extends LivekitTransport>(
t1: T | null,
t2: T | null,
): boolean {
if (t1 && t2) return t1.livekit_service_url === t2.livekit_service_url;
// In case we have different lk rooms in the same SFU (depends on the livekit authorization service)

View File

@@ -24,6 +24,7 @@ import {
mockCallMembership,
mockComputeLivekitParticipantIdentity$,
mockMediaDevices,
ownMemberMock,
withTestScheduler,
} from "../../../utils/test.ts";
import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
@@ -128,6 +129,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
connectionFactory: ecConnectionFactory,
inputTransports$: membershipsAndTransports.transports$,
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
const matrixLivekitItems$ = createMatrixLivekitMembers$({

View File

@@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details.
import {
type CallMembership,
isLivekitTransport,
type LivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
} from "matrix-js-sdk/lib/matrixrtc";
@@ -21,15 +20,18 @@ import {
type ObservableScope,
} from "./ObservableScope";
import { type Behavior } from "./Behavior";
import { type LivekitTransportWithVersion } from "./CallViewModel/remoteMembers/ConnectionManager";
export const membershipsAndTransports$ = (
scope: ObservableScope,
memberships$: Behavior<Epoch<CallMembership[]>>,
): {
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<
{ membership: CallMembership; transport?: LivekitTransportWithVersion }[]
>
>;
transports$: Behavior<Epoch<LivekitTransport[]>>;
transports$: Behavior<Epoch<LivekitTransportWithVersion[]>>;
} => {
/**
* Lists the transports used by ourselves, plus all other MatrixRTC session
@@ -47,7 +49,12 @@ export const membershipsAndTransports$ = (
const transport = membership.getTransport(oldestMembership);
return {
membership,
transport: isLivekitTransport(transport) ? transport : undefined,
transport: isLivekitTransport(transport)
? {
...transport,
useMatrix2: membership.kind === "rtc",
}
: undefined,
};
});
}),

View File

@@ -25,7 +25,6 @@ import {
import {
CallMembership,
type LivekitFocusSelection,
type LivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
type MatrixRTCSessionEventHandlerMap,
@@ -67,6 +66,7 @@ import { type MediaDevices } from "../state/MediaDevices";
import { type Behavior, constant } from "../state/Behavior";
import { ObservableScope } from "../state/ObservableScope";
import { MuteStates } from "../state/MuteStates";
import { type LivekitTransportWithVersion } from "../state/CallViewModel/remoteMembers/ConnectionManager";
export function withFakeTimers(continuation: () => void): void {
vi.useFakeTimers();
@@ -197,10 +197,11 @@ export function mockEmitter<T>(): EmitterMock<T> {
};
}
export const exampleTransport: LivekitTransport = {
export const exampleTransport: LivekitTransportWithVersion = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
livekit_alias: "!alias:example.org",
useMatrix2: false,
};
export function mockCallMembership(
@@ -256,6 +257,11 @@ export function mockRtcMembership(
return cms;
}
export const ownMemberMock: CallMembershipIdentityParts = {
userId: "@alice:example.org",
deviceId: "DEVICE",
memberId: "@alice:example.org:DEVICE",
};
// Maybe it'd be good to move this to matrix-js-sdk? Our testing needs are
// rather simple, but if one util to mock a member is good enough for us, maybe
// it's useful for matrix-js-sdk consumers in general.

View File

@@ -10338,9 +10338,9 @@ __metadata:
languageName: node
linkType: hard
"matrix-js-sdk@github:matrix-org/matrix-js-sdk#head=toger5/use-membershipID-for-session-state-events&commit=f5f1b8efb46b3d55a7eebfabb4a61496640b8b00":
version: 39.3.0
resolution: "matrix-js-sdk@https://github.com/matrix-org/matrix-js-sdk.git#commit=f5f1b8efb46b3d55a7eebfabb4a61496640b8b00"
"matrix-js-sdk@portal:/Users/timo/Projects/matrix-js-sdk::locator=element-call%40workspace%3A.":
version: 0.0.0-use.local
resolution: "matrix-js-sdk@portal:/Users/timo/Projects/matrix-js-sdk::locator=element-call%40workspace%3A."
dependencies:
"@babel/runtime": "npm:^7.12.5"
"@matrix-org/matrix-sdk-crypto-wasm": "npm:^16.0.0"
@@ -10356,9 +10356,8 @@ __metadata:
sdp-transform: "npm:^3.0.0"
unhomoglyph: "npm:^1.0.6"
uuid: "npm:13"
checksum: 10c0/9607b0c063c873a24c1a2d05cc7500d60c32556ec82b666ebaae5c5e829faf5bb7639780efddea7211e6b9873098bd53b97656f041e932e8b0de0c208ccabbff
languageName: node
linkType: hard
linkType: soft
"matrix-widget-api@npm:^1.14.0":
version: 1.15.0