This commit is contained in:
Robin
2026-10-03 01:27:05 +02:00
parent 0030371ea7
commit 8a06976494
23 changed files with 232 additions and 297 deletions
@@ -31,7 +31,7 @@ import {
type CallMembership,
type IRTCNotificationContent,
MatrixRTCSessionEvent,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { deepCompare } from "matrix-js-sdk/lib/utils";
@@ -97,7 +97,7 @@ vi.mock(
"../state/CallViewModel/localMember/localTransport",
async (importOriginal) => ({
...(await importOriginal()),
makeTransport: async (): Promise<LivekitTransport> =>
makeTransport: async (): Promise<UnstableLivekitTransport> =>
Promise.resolve(exampleTransport),
}),
);
+5 -7
View File
@@ -43,7 +43,7 @@ import {
import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
import {
MembershipManagerEvent,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
type RTCCallIntent,
type RTCNotificationType,
@@ -568,6 +568,8 @@ export function createCallViewModel$(
// membership advertises `${userId}:${deviceId}` as its `membershipID`
// instead, so a uuid there names a member no peer can resolve.
memberId:
// TODO: Make this conditional on the transport instead
// or else avoid using it in key messages
matrixRTCMode === MatrixRTCMode.Matrix_2_0
? uuidv4()
: `${userId}:${deviceId}`,
@@ -580,7 +582,6 @@ export function createCallViewModel$(
ownMembershipIdentity,
client,
roomId: matrixRoom.roomId,
matrixRTCMode,
}),
);
@@ -624,7 +625,7 @@ export function createCallViewModel$(
),
muteStates,
joinMatrixRTC: (
transport: LivekitTransport,
transport: UnstableLivekitTransport,
delayedLeaveTimings: ResolvedDelayedLeaveTimings,
) => {
return enterRTCSession(
@@ -646,9 +647,7 @@ export function createCallViewModel$(
mediaDevices,
muteStates,
trackProcessorState$,
logger.getChild(
"[Publisher " + connection.transport.livekit_service_url + "]",
),
logger.getChild(`[Publisher ${JSON.stringify(connection.transport)}]`),
controlledAudioDevices,
);
},
@@ -671,7 +670,6 @@ export function createCallViewModel$(
).pipe(map(([delayId]) => delayId ?? null)),
matrixRTCSession.delayId ?? null,
),
matrixRTCMode,
logger: logger.getChild(`[${Date.now()}]`),
});
@@ -8,7 +8,7 @@ Please see LICENSE in the repository root for full details.
import {
Status as RTCMemberStatus,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
} from "matrix-js-sdk/lib/matrixrtc";
import {
@@ -121,7 +121,7 @@ const mockedClient = {
};
describe("enterRTCSession", () => {
const transport: LivekitTransport = {
const transport: UnstableLivekitTransport = {
livekit_service_url: "http://my-livekit-service-url.com",
type: "livekit",
};
@@ -252,7 +252,7 @@ describe("LocalMembership", () => {
it("throws error on missing RTC config error", () => {
withTestScheduler(({ scope, hot, expectObservable }) => {
const localTransport$ = scope.behavior<null | LivekitTransport>(
const localTransport$ = scope.behavior<null | UnstableLivekitTransport>(
hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")),
null,
);
@@ -317,7 +317,7 @@ describe("LocalMembership", () => {
const mockTransportConfig = {
livekit_service_url: "a",
} as LivekitTransport;
} as UnstableLivekitTransport;
const mockTransport = {
transport: mockTransportConfig,
@@ -17,7 +17,7 @@ import {
import { observeParticipantEvents } from "@livekit/components-core";
import {
Status as RTCSessionStatus,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
type RTCCallIntent,
type RTCNotificationType,
@@ -82,7 +82,8 @@ import { type LocalTransport } from "./LocalTransport.ts";
import {
type ClientDelegationParts,
delegateDelayedLeave,
} from "../../../livekit/auth/index.ts";
type ClientOpenIDParts,
} from "../../../livekit/auth";
export enum TransportState {
/** Not even a transport is available to the LocalMembership */
@@ -145,14 +146,14 @@ interface Props {
connectionManager: IConnectionManager;
createPublisherFactory: (connection: Connection) => Publisher;
joinMatrixRTC: (
transport: LivekitTransport,
transport: UnstableLivekitTransport,
delayedLeaveTimings: ResolvedDelayedLeaveTimings,
) => void;
homeserverConnected: HomeserverConnected;
roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
localTransport$: Observable<LocalTransport>;
client: ClientDelegationParts;
client: ClientDelegationParts & ClientOpenIDParts;
matrixRTCSession: Pick<MatrixRTCSession, "updateCallIntent" | "leave">;
/** Whether to hide the screen-sharing button. */
hideScreensharing: boolean;
@@ -160,7 +161,6 @@ interface Props {
hostBridge: HostBridge;
baseUrl: string;
delayId$: Behavior<string | null>;
matrixRTCMode: MatrixRTCMode;
logger: Logger;
}
@@ -208,7 +208,6 @@ export const createLocalMembership$ = ({
hostBridge,
ownMembershipIdentity,
delayId$,
matrixRTCMode,
}: Props): {
/**
* This request to start audio and video tracks.
@@ -714,7 +713,6 @@ export const createLocalMembership$ = ({
transport: joinParams.transport,
roomId,
delayId,
matrixRTCMode,
logger,
});
logger.info("Delayed leave successfully delegated");
@@ -953,7 +951,7 @@ interface EnterRTCSessionOptions {
*
* @param rtcSession - The MatrixRTCSession to join.
* @param ownMembershipIdentity - Options for entering the RTC session.
* @param transport - The LivekitTransport to use for this session.
* @param transport - The UnstableLivekitTransport to use for this session.
* @param options - `encryptMedia`: Whether to encrypt media. `matrixRTCMode`: The
* Matrix RTC mode to use. `delayedLeaveTimings`: The preferred timings for
* delayed leave events. `sendNotificationType`: Whether and what kind of
@@ -964,7 +962,7 @@ interface EnterRTCSessionOptions {
export function enterRTCSession(
rtcSession: MatrixRTCSession,
ownMembershipIdentity: CallMembershipIdentityParts,
transport: LivekitTransport,
transport: UnstableLivekitTransport,
{
encryptMedia,
matrixRTCMode,
@@ -5,9 +5,9 @@ SPDX-License-IdFentifier: 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 { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { Config } from "../../../config/Config.ts";
@@ -16,31 +16,31 @@ import {
MatrixRTCTransportMissingError,
NoMatrix2AuthorizationService,
} from "../../../utils/errors.ts";
import {
getSFUConfigWithOpenID,
type SFUConfig,
type OpenIDClientParts,
} from "../../../livekit/openIDSFU.ts";
import { customLivekitUrl } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type MatrixRTCMode } from "../../../config/ConfigOptions.ts";
import {
getSFUConfig,
type ClientGetTokenParts,
type ClientOpenIDParts,
type SFUConfig,
} from "../../../livekit/auth";
interface Props {
ownMembershipIdentity: CallMembershipIdentityParts;
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts;
ClientGetTokenParts &
ClientOpenIDParts;
// Used by the jwt service to create the livekit room and compute the livekit alias.
roomId: string;
matrixRTCMode: MatrixRTCMode;
}
export interface LocalTransport {
transport: LivekitTransport;
transport: UnstableLivekitTransport;
sfuConfig: SFUConfig;
}
export function isLocalTransport(
obj: LivekitTransport | LocalTransport,
obj: UnstableLivekitTransport | LocalTransport,
): obj is LocalTransport {
return "transport" in obj && "sfuConfig" in obj;
}
@@ -58,7 +58,6 @@ export async function getLocalTransport({
ownMembershipIdentity,
client,
roomId,
matrixRTCMode,
}: Props): Promise<LocalTransport> {
const logger = rootLogger.getChild("[LocalTransport]");
const discovery = new RtcTransportAutoDiscovery({
@@ -69,7 +68,7 @@ export async function getLocalTransport({
const customUrl = customLivekitUrl.value$.value;
// Respect the user's custom URL, if set
const transport: LivekitTransport | null = customUrl
const transport: UnstableLivekitTransport | null = customUrl
? { type: "livekit", livekit_service_url: customUrl }
: await discovery.discoverPreferredTransport();
@@ -77,60 +76,26 @@ export async function getLocalTransport({
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
try {
return await doOpenIdAndJWTFromUrl(
return {
transport,
matrixRTCMode,
ownMembershipIdentity,
roomId,
client,
logger,
);
sfuConfig: await getSFUConfig({
client,
membership: ownMembershipIdentity,
transport,
roomId,
role: "publisher",
logger,
}),
};
} catch (e) {
logger.error(
`Failed to authenticate to transport ${transport.livekit_service_url}`,
`Failed to authenticate to transport ${JSON.stringify(transport)}`,
e,
);
throw mapAuthErrorToUserFriendlyError(e);
}
}
/**
* Utility to ensure the user can authenticate with the SFU.
* We will call `getSFUConfigWithOpenID` once per transport here as it's our
* only mechanism of validation. This means we will also ask the
* homeserver for a OpenID token a few times. Since OpenID tokens are single
* use we don't want to risk any issues by re-using a token.
*
* @param transport The transport to authenticate with.
* @param matrixRTCMode Whether to force the JWT endpoint to be used.
* @param membership The identity of the local member.
* @param roomId The room ID to use for the JWT.
* @param client The client to use for the OpenID token.
*
* @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService
*/
async function doOpenIdAndJWTFromUrl(
transport: LivekitTransport,
matrixRTCMode: MatrixRTCMode,
membership: CallMembershipIdentityParts,
roomId: string,
client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts,
logger?: Logger,
): Promise<LocalTransport> {
const sfuConfig = await getSFUConfigWithOpenID(
client,
membership,
transport.livekit_service_url,
roomId,
{ matrixRTCMode },
logger,
);
return {
transport,
sfuConfig,
};
}
function mapAuthErrorToUserFriendlyError(e: unknown): Error {
if (
e instanceof FailToGetOpenIdToken ||
@@ -16,7 +16,7 @@ import {
import { MatrixError } from "matrix-js-sdk";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import {
type LivekitTransport,
type UnstableLivekitTransport,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
@@ -28,12 +28,12 @@ import {
type DiscoveryClient = RtcTransportAutoDiscoveryProps["client"];
const backendTransport: LivekitTransport = {
const backendTransport: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://backend.example.org",
};
const configTransport: LivekitTransport = {
const configTransport: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://config.example.org",
};
@@ -5,8 +5,8 @@ SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import {
isLivekitTransport,
type LivekitTransport,
isUnstableLivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger";
@@ -40,7 +40,7 @@ export class RtcTransportAutoDiscovery {
this.logger = logger.getChild("[RtcTransportAutoDiscovery]");
}
public async discoverPreferredTransport(): Promise<LivekitTransport | null> {
public async discoverPreferredTransport(): Promise<UnstableLivekitTransport | null> {
// 1) backend transports
const backendTransport = await this.tryBackendTransports();
if (backendTransport) {
@@ -67,7 +67,7 @@ export class RtcTransportAutoDiscovery {
* This will not throw errors, but instead just log them and return null if the expected config is not found or malformed.
* @private
*/
private async tryBackendTransports(): Promise<LivekitTransport | null> {
private async tryBackendTransports(): Promise<UnstableLivekitTransport | null> {
const client = this.client;
// MSC4143: Attempt to fetch transports from backend.
this.logger.info("First try to use getRTCTransports end point ...");
@@ -75,7 +75,7 @@ export class RtcTransportAutoDiscovery {
const transportList = await doNetworkOperationWithRetry(async () =>
client._unstable_getRTCTransports(),
);
const first = transportList.find(isLivekitTransport);
const first = transportList.find(isUnstableLivekitTransport);
if (first) {
return first;
} else {
@@ -90,7 +90,7 @@ export class RtcTransportAutoDiscovery {
return null;
}
private tryConfigTransport(): LivekitTransport | null {
private tryConfigTransport(): UnstableLivekitTransport | null {
const url = this.resolvedConfig.livekit?.livekit_service_url;
if (url) {
return {
@@ -30,7 +30,7 @@ import fetchMock from "fetch-mock";
import EventEmitter from "events";
import { type IOpenIDToken } from "matrix-js-sdk";
import { logger, type Logger } from "matrix-js-sdk/lib/logger";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
Connection,
@@ -38,7 +38,7 @@ import {
type ConnectionOpts,
} from "./Connection.ts";
import { ObservableScope } from "../../ObservableScope.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import { type ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
import {
ElementCallError,
FailToGetOpenIdToken,
@@ -48,7 +48,7 @@ import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts";
let testScope: ObservableScope;
let client: MockedObject<OpenIDClientParts>;
let client: MockedObject<ClientOpenIDParts>;
let fakeLivekitRoom: MockedObject<LivekitRoom>;
@@ -57,14 +57,14 @@ let fakeLocalParticipant: MockedObject<LocalParticipant>;
const ROOM_ID = "!roomID:example.org";
const livekitFocus: LivekitTransport = {
const livekitFocus: UnstableLivekitTransport = {
livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
type: "livekit",
};
function setupTest(): void {
testScope = new ObservableScope();
client = vi.mocked<OpenIDClientParts>({
client = vi.mocked<ClientOpenIDParts>({
getOpenIdToken: vi.fn().mockResolvedValue({
access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
token_type: "Bearer",
@@ -72,7 +72,7 @@ function setupTest(): void {
expires_in: 3600,
}),
getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
} as unknown as OpenIDClientParts);
} as unknown as ClientOpenIDParts);
localParticipantEventEmiter = new EventEmitter();
@@ -22,16 +22,11 @@ import {
type Track,
type TrackPublication,
} from "livekit-client";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } 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,
type OpenIDClientParts,
type SFUConfig,
} from "../../../livekit/openIDSFU.ts";
import { type Behavior } from "../../Behavior.ts";
import { type ObservableScope } from "../../ObservableScope.ts";
import {
@@ -42,6 +37,12 @@ import {
SFURoomCreationRestrictedError,
UnknownCallError,
} from "../../../utils/errors.ts";
import {
type ClientGetTokenParts,
getSFUConfig,
type ClientOpenIDParts,
type SFUConfig,
} from "../../../livekit/auth";
export interface ConnectionOpts {
/**
@@ -52,9 +53,9 @@ export interface ConnectionOpts {
/** The identity parts to use on this connection */
ownMembershipIdentity: CallMembershipIdentityParts;
/** The media transport to connect to. */
transport: LivekitTransport;
transport: UnstableLivekitTransport;
/** The Matrix client to use for OpenID and SFU config requests. */
client: OpenIDClientParts;
client: ClientGetTokenParts & ClientOpenIDParts;
/** The room ID this connection is associated with. */
roomId: string;
/** The observable scope to use for this connection. */
@@ -107,7 +108,7 @@ export class Connection {
/**
* The media transport to connect to.
*/
public readonly transport: LivekitTransport;
public readonly transport: UnstableLivekitTransport;
public readonly livekitRoom: LivekitRoom;
@@ -137,7 +138,7 @@ export class Connection {
protected stopped = false;
// TODO: can we just keep the ConnectionOpts object instead of spreading?
private readonly client: OpenIDClientParts;
private readonly client: ClientGetTokenParts & ClientOpenIDParts;
private readonly roomId: string;
private readonly logger: Logger;
private readonly ownMembershipIdentity: CallMembershipIdentityParts;
@@ -154,10 +155,10 @@ export class Connection {
this.existingSFUConfig = opts.existingSFUConfig;
this.roomId = opts.roomId;
this.logger = logger.getChild(
"[Connection " + opts.transport.livekit_service_url + "]",
`[Connection ${JSON.stringify(opts.transport)}]`,
);
this.logger.info(
`constructor: ${opts.transport.livekit_service_url} roomId: ${this.roomId} withSfuConfig?: ${opts.existingSFUConfig ? JSON.stringify(opts.existingSFUConfig) : "undefined"}`,
`constructor - roomId: ${this.roomId} withSfuConfig?: ${opts.existingSFUConfig ? JSON.stringify(opts.existingSFUConfig) : "undefined"}`,
);
const { transport, client, scope } = opts;
@@ -311,9 +312,7 @@ export class Connection {
this.existingSFUConfig ??
(await this.getSFUConfigForRemoteConnection());
this.logger.debug(
"Starting Connection to: ",
this.transport.livekit_service_url,
"jwt: ",
"Starting Connection - jwt: ",
jwt,
"wss: ",
url,
@@ -398,15 +397,14 @@ export class Connection {
protected async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
// This will only be called for sfu's where we do not publish ourselves.
// For the local connection we will use the existingJwtTokenData
return await getSFUConfigWithOpenID(
this.client,
this.ownMembershipIdentity,
this.transport.livekit_service_url,
this.roomId,
// dont pass any custom opts for the subscribe only connections
{},
this.logger,
);
return await getSFUConfig({
client: this.client,
membership: this.ownMembershipIdentity,
transport: this.transport,
roomId: this.roomId,
role: "subscriber",
logger: this.logger,
});
}
/**
@@ -416,17 +414,13 @@ export class Connection {
* If the connection is already stopped, this is a no-op.
*/
public async stop(): Promise<void> {
this.logger.debug(
`stop: disconnecing from lk room ${this.transport.livekit_service_url}`,
);
this.logger.debug("stop: disconnecing from lk room");
if (this.stopped) return;
// Mark as stopped before disconnecting so that a connect() aborted by the
// disconnect sees the flag and does not report the abort as an error.
this.stopped = true;
await this.livekitRoom.disconnect();
this._state$.next(ConnectionState.Stopped);
this.logger.debug(
`stop: DONE disconnecing from lk room ${this.transport.livekit_service_url}`,
);
this.logger.debug("stop: DONE disconnecing from lk room");
}
}
@@ -16,12 +16,12 @@ import { logger, type Logger } from "matrix-js-sdk/lib/logger";
// imported as inline to support worker when loaded from a cdn (cross domain)
import E2EEWorker from "livekit-client/e2ee-worker?worker&inline";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type ObservableScope } from "../../ObservableScope.ts";
import { Connection } from "./Connection.ts";
import type {
OpenIDClientParts,
ClientOpenIDParts,
SFUConfig,
} from "../../../livekit/openIDSFU.ts";
import type { MediaDevices } from "../../MediaDevices.ts";
@@ -44,7 +44,7 @@ import {
export interface ConnectionFactory {
createConnection(
scope: ObservableScope,
transport: LivekitTransport,
transport: UnstableLivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
@@ -66,7 +66,7 @@ export class ECConnectionFactory implements ConnectionFactory {
* @param livekitRoomFactory - Optional factory function (for testing) to create LivekitRoom instances. If not provided, a default factory is used.
*/
public constructor(
private client: OpenIDClientParts,
private client: ClientOpenIDParts,
private readonly roomId: string,
private devices: MediaDevices,
private processorState$: Behavior<ProcessorState>,
@@ -103,7 +103,7 @@ export class ECConnectionFactory implements ConnectionFactory {
*/
public createConnection(
scope: ObservableScope,
transport: LivekitTransport,
transport: UnstableLivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
@@ -7,7 +7,7 @@ Please see LICENSE in the repository root for full details.
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { BehaviorSubject, NEVER } from "rxjs";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type RemoteParticipant } from "livekit-client";
import { logger } from "matrix-js-sdk/lib/logger";
@@ -19,17 +19,17 @@ import {
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type Behavior } from "../../Behavior.ts";
// Some test constants
const TRANSPORT_1: LivekitTransport = {
const TRANSPORT_1: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const TRANSPORT_2: LivekitTransport = {
const TRANSPORT_2: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
};
@@ -47,7 +47,7 @@ beforeEach(() => {
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
(scope: ObservableScope, transport: UnstableLivekitTransport) => {
const mockConnection = {
transport,
remoteParticipants$: new BehaviorSubject([]),
@@ -94,13 +94,13 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalled();
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
expect(conn2!.start).toHaveBeenCalled();
@@ -141,12 +141,12 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalledOnce();
@@ -187,7 +187,7 @@ describe("connections$ stream", () => {
expect(connections.length).toBe(1);
// The second connection should have been stopped has it is no longer needed.
const connection2 = allCreatedConnections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(connection2).toBeDefined();
expect(connection2!.stop).toHaveBeenCalled();
@@ -225,7 +225,7 @@ describe("connectionManagerData$ stream", () => {
// Used in test to control fake connections' remoteParticipants$ streams
let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
function keyForTransport(transport: LivekitTransport): string {
function keyForTransport(transport: UnstableLivekitTransport): string {
return `${transport.livekit_service_url}`;
}
@@ -233,7 +233,7 @@ describe("connectionManagerData$ stream", () => {
fakeRemoteParticipantsStreams = new Map();
function getRemoteParticipantsFor(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): Behavior<RemoteParticipant[]> {
return (
fakeRemoteParticipantsStreams.get(keyForTransport(transport)) ??
@@ -245,7 +245,7 @@ describe("connectionManagerData$ stream", () => {
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
(scope: ObservableScope, transport: UnstableLivekitTransport) => {
const fakeRemoteParticipants$ = new BehaviorSubject<
RemoteParticipant[]
>([]);
@@ -6,7 +6,7 @@ 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 { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
combineLatest,
map,
@@ -25,7 +25,7 @@ import { type Behavior } from "../../Behavior.ts";
import { type Connection } from "./Connection.ts";
import { Epoch, type ObservableScope } from "../../ObservableScope.ts";
import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import {
isLocalTransport,
@@ -56,7 +56,7 @@ export class ConnectionManagerData {
}
}
private getKey(transport: LivekitTransport): string {
private getKey(transport: UnstableLivekitTransport): string {
// This is enough as a key because the ConnectionManager is already scoped by room.
// We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`.
return transport.livekit_service_url;
@@ -67,13 +67,13 @@ export class ConnectionManagerData {
}
public getConnectionForTransport(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): Connection | null {
return this.store.get(this.getKey(transport))?.connection ?? null;
}
public getParticipantsForTransport(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): RemoteParticipant[] {
const key = this.getKey(transport);
const existing = this.store.get(key);
@@ -88,7 +88,7 @@ interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
localTransport$: Observable<LocalTransport>;
remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
remoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
logger: Logger;
ownMembershipIdentity: CallMembershipIdentityParts;
@@ -147,7 +147,7 @@ export function createConnectionManager$({
* externally this is modified via `registerTransports()`.
*/
const localAndRemoteTransports$: Behavior<
Epoch<(LivekitTransport | LocalTransport)[]>
Epoch<(UnstableLivekitTransport | LocalTransport)[]>
> = scope.behavior(
combineLatest([localTransportAsArray$, remoteTransports$]).pipe(
// Combine local and remote transports into one transport array
@@ -157,7 +157,7 @@ export function createConnectionManager$({
const remoteWithoutLocal = dedupedRemote.filter(
(transport) =>
!localTransportAsArray.find((l) =>
areLivekitTransportsEqual(l.transport, transport),
areUnstableLivekitTransportsEqual(l.transport, transport),
),
);
logger.debug(
@@ -271,11 +271,11 @@ export function createConnectionManager$({
return { connectionManagerData$ };
}
function removeDuplicateTransports<T extends LivekitTransport>(
function removeDuplicateTransports<T extends UnstableLivekitTransport>(
transports: T[],
): T[] {
return transports.reduce((acc, transport) => {
if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
if (!acc.some((t) => areUnstableLivekitTransportsEqual(t, transport)))
acc.push(transport);
return acc;
}, [] as T[]);
@@ -14,7 +14,7 @@ import EventEmitter from "events";
import { ObservableScope } from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import type { OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import type { ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
import {
exampleTransport,
mockMediaDevices,
@@ -51,7 +51,7 @@ vi.mock("livekit-client", async (importOriginal) => {
});
let testScope: ObservableScope;
let mockClient: OpenIDClientParts;
let mockClient: ClientOpenIDParts;
beforeEach(() => {
testScope = new ObservableScope();
@@ -9,7 +9,7 @@ import { describe, test, expect, beforeEach, afterEach, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, combineLatest, map, type Observable } from "rxjs";
@@ -39,12 +39,12 @@ let testScope: ObservableScope;
const fallbackMemberId = (userId: string, deviceId: string): string =>
`${userId}:${deviceId}`;
const transportA: LivekitTransport = {
const transportA: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const transportB: LivekitTransport = {
const transportB: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
};
@@ -117,7 +117,7 @@ test("should signal participant not yet connected to livekit", async () => {
function createEpochedMemberships$(m$: Observable<CallMembership[]>): {
memberships$: Observable<Epoch<CallMembership[]>>;
membershipsWithTransport$: Observable<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
>;
} {
const memberships$ = m$.pipe(trackEpoch());
@@ -128,7 +128,7 @@ function createEpochedMemberships$(m$: Observable<CallMembership[]>): {
return {
membership: m,
transport:
tr?.type === "livekit" ? (tr as LivekitTransport) : undefined,
tr?.type === "livekit" ? (tr as UnstableLivekitTransport) : undefined,
};
});
}),
@@ -8,7 +8,7 @@ Please see LICENSE in the repository root for full details.
import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, filter, map } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
@@ -60,7 +60,7 @@ export interface RemoteMatrixLivekitMember extends MatrixLivekitMember {
interface Props {
scope: ObservableScope;
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
>;
connectionManager: IConnectionManager;
localUser: { deviceId: string; userId: string };
@@ -165,7 +165,7 @@ export function createRemoteMatrixLivekitMembers$({
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
// TODO add this to the JS-SDK
export function areLivekitTransportsEqual<T extends LivekitTransport>(
export function areUnstableLivekitTransportsEqual<T extends UnstableLivekitTransport>(
t1: T | null,
t2: T | null,
): boolean {
@@ -10,7 +10,7 @@ import { BehaviorSubject, NEVER } from "rxjs";
import { type Room as LivekitRoom } from "livekit-client";
import EventEmitter from "events";
import fetchMock from "fetch-mock";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { logger } from "matrix-js-sdk/lib/logger";
import {
@@ -19,7 +19,7 @@ import {
trackEpoch,
} from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import { type ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
import {
mockMediaDevices,
mockRtcMembership,
@@ -28,7 +28,7 @@ import {
} from "../../../utils/test.ts";
import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
import {
areLivekitTransportsEqual,
areUnstableLivekitTransportsEqual,
createRemoteMatrixLivekitMembers$,
type RemoteMatrixLivekitMember,
} from "./MatrixLivekitMembers.ts";
@@ -40,7 +40,7 @@ import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
let testScope: ObservableScope;
let ecConnectionFactory: ECConnectionFactory;
let mockClient: OpenIDClientParts;
let mockClient: ClientOpenIDParts;
let lkRoomFactory: () => LivekitRoom;
const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
@@ -147,9 +147,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
});
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((co) =>
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
co.transport,
bobMembership.transports[0]! as LivekitTransport,
bobMembership.transports[0]! as UnstableLivekitTransport,
),
),
});
@@ -184,9 +184,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((connection) => {
expect(
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
connection.transport,
carlMembership.transports[0]! as LivekitTransport,
carlMembership.transports[0]! as UnstableLivekitTransport,
),
).toBe(true);
return true;
@@ -214,9 +214,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((connection) => {
expect(
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
connection.transport,
daveMembership.transports[0]! as LivekitTransport,
daveMembership.transports[0]! as UnstableLivekitTransport,
),
).toBe(true);
return true;
+7 -7
View File
@@ -7,10 +7,10 @@ Please see LICENSE in the repository root for full details.
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
isLivekitTransport,
isUnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { fromEvent } from "rxjs";
@@ -31,15 +31,15 @@ export const membershipsAndTransports$ = (
memberships$: Behavior<Epoch<CallMembership[]>>,
): {
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
>;
transports$: Behavior<Epoch<LivekitTransport[]>>;
transports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
} => {
const membershipsWithTransport$: Behavior<
Epoch<
{
membership: CallMembership;
transport: LivekitTransport | undefined;
transport: UnstableLivekitTransport | undefined;
}[]
>
> = scope.behavior(
@@ -49,14 +49,14 @@ export const membershipsAndTransports$ = (
const transport = membership.getTransport();
return {
membership,
transport: isLivekitTransport(transport) ? transport : undefined,
transport: isUnstableLivekitTransport(transport) ? transport : undefined,
};
});
}),
),
);
const transports$: Behavior<Epoch<LivekitTransport[]>> = scope.behavior(
const transports$: Behavior<Epoch<UnstableLivekitTransport[]>> = scope.behavior(
membershipsWithTransport$.pipe(
mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
),