Account for server name when identifying transports

This commit is contained in:
Robin
2026-10-09 01:43:25 +02:00
parent 4c96c99c91
commit 033f79975a
20 changed files with 289 additions and 143 deletions
+6
View File
@@ -29,6 +29,10 @@ export interface GetSFUConfigParams {
* The WebSocket URL of the SFU for which we wish to get an access token.
*/
url: string;
/**
* The name of the homeserver to which the SFU belongs.
*/
serverName: string;
/**
* The ID of the Matrix room in which the session takes place.
*/
@@ -47,12 +51,14 @@ export async function getSFUConfig({
client,
membership,
url,
serverName,
roomId,
slotId,
}: GetSFUConfigParams): Promise<SFUConfig> {
const res = await doNetworkOperationWithRetry(async () =>
client._unstable_getLivekitToken({
url,
server_name: serverName,
room_id: roomId,
slot_id: slotId,
member_id: membership.memberId,
+6
View File
@@ -50,6 +50,10 @@ export interface GetSFUConfigParams {
* The transport for which we wish to get an access token.
*/
transport: UnstableLivekitTransport;
/**
* The name of the homeserver to which the {@link transport} belongs.
*/
serverName: string;
/**
* The ID of the Matrix room in which the session takes place.
*/
@@ -72,6 +76,7 @@ export async function getSFUConfig({
client,
membership,
transport,
serverName,
roomId,
slotId,
role,
@@ -83,6 +88,7 @@ export async function getSFUConfig({
client,
membership,
url: transport.url,
serverName,
roomId,
slotId,
});
+11
View File
@@ -5,6 +5,17 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
/**
* Data identifying a transport and the homeserver to which it belongs, so
* consumers know how to reach it.
*/
export interface TransportLocator {
transport: UnstableLivekitTransport;
serverName: string;
}
/**
* Configuration and access tokens provided by the SFU on successful authentication.
*/
+1
View File
@@ -526,6 +526,7 @@ export const DeveloperSettingsTab: FC<Props> = ({
client,
membership: { userId, deviceId, memberId: "" },
transport,
serverName: client.getDomain()!,
roomId,
slotId: "org.example.fake-slot#room",
role: "publisher",
+2 -1
View File
@@ -163,6 +163,7 @@ import {
type RingingMediaViewModel,
} from "../media/RingingMediaViewModel.ts";
import { type GridTileViewModel } from "../TileViewModel.ts";
import { type TransportLocator } from "../../livekit/auth";
//TODO
// Larger rename
@@ -215,7 +216,7 @@ export interface CallViewModelOptions {
*/
windowSize$: Behavior<{ width: number; height: number }>;
/** Optional value overriding the local transport, for testing purposes. */
localTransport?: UnstableLivekitTransport;
localTransport?: TransportLocator;
/** Optional value overriding the connection factory, for testing purposes. */
connectionFactory?: ConnectionFactory;
/** The version & compatibility mode of MatrixRTC that we should use. */
@@ -30,7 +30,7 @@ import {
type CallViewModelOptions,
} from "./CallViewModel";
import {
exampleTransport,
exampleTransportLocator,
mockConfig,
MockConnection,
mockLivekitRoom,
@@ -208,12 +208,13 @@ export function withCallViewModel(mode: MatrixRTCMode) {
livekitRoomFactory,
connectionState$,
windowSize$,
localTransport: exampleTransport,
localTransport: exampleTransportLocator,
connectionFactory: {
createConnection(
scope,
role,
transport,
serverName,
ownMembershipIdentity,
logger,
) {
@@ -222,6 +223,7 @@ export function withCallViewModel(mode: MatrixRTCMode) {
scope,
role,
transport,
serverName,
ownMembershipIdentity,
client: room.client,
roomId: room.roomId,
@@ -305,7 +305,7 @@ describe("LocalMembership", () => {
leave: vi.fn(),
},
connectionManager: mockConnectionManager,
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
const expextedLog =
"'not connected yet' while updating the call intent (this is expected on startup)";
@@ -321,7 +321,12 @@ describe("LocalMembership", () => {
type: "livekit",
url: "https://sfu.example.org",
livekit_service_url: "https://jwt.example.org",
} as UnstableLivekitTransport;
};
const mockTransportLocator: livekitAuth.TransportLocator = {
transport: mockTransport,
serverName: "example.org",
};
const connectionTransportAConnected = {
livekitRoom: mockLivekitRoom({
@@ -375,7 +380,7 @@ describe("LocalMembership", () => {
),
},
joinMatrixRTC,
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
delayId$,
});
@@ -452,7 +457,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
await flushPromises();
expect(publisherFactory).toHaveBeenCalledOnce();
@@ -474,7 +479,7 @@ describe("LocalMembership", () => {
it("tracks livekit state correctly", async () => {
const scope = new ObservableScope();
const connectionManagerData = new ConnectionManagerData();
const localTransport$ = new Subject<UnstableLivekitTransport>();
const localTransport$ = new Subject<livekitAuth.TransportLocator>();
const connectionManagerData$ = new BehaviorSubject(
new Epoch(connectionManagerData),
@@ -522,7 +527,7 @@ describe("LocalMembership", () => {
expect(localMembership.localMemberState$.value).toStrictEqual(
TransportState.Waiting,
);
localTransport$.next(mockTransport);
localTransport$.next(mockTransportLocator);
await flushPromises();
expect(localMembership.localMemberState$.value).toStrictEqual({
matrix: RTCMemberStatus.Connected,
@@ -547,7 +552,7 @@ describe("LocalMembership", () => {
});
(
connectionManagerData2.getConnectionForTransport(mockTransport)!
connectionManagerData2.getConnectionForTransport(mockTransportLocator)!
.state$ as BehaviorSubject<ConnectionState>
).next(ConnectionState.LivekitConnected);
expect(localMembership.localMemberState$.value).toStrictEqual({
@@ -650,7 +655,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
await flushPromises();
@@ -687,7 +692,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
await flushPromises();
@@ -735,7 +740,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
await flushPromises();
@@ -776,7 +781,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
await flushPromises();
@@ -842,7 +847,7 @@ describe("LocalMembership", () => {
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: constant(mockTransport),
localTransport$: constant(mockTransportLocator),
});
return { scope, localMembership };
};
@@ -82,6 +82,7 @@ import {
type ClientDelegationParts,
delegateDelayedLeave,
type ClientOpenIDParts,
type TransportLocator,
} from "../../../livekit/auth";
export enum TransportState {
@@ -151,7 +152,7 @@ interface Props {
homeserverConnected: HomeserverConnected;
roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
localTransport$: Observable<UnstableLivekitTransport>;
localTransport$: Observable<TransportLocator>;
client: ClientDelegationParts & ClientOpenIDParts;
matrixRTCSession: Pick<
MatrixRTCSession,
@@ -310,7 +311,7 @@ export const createLocalMembership$ = ({
// to whether delayed event delegation is supported
const joinParams$ = scope.behavior(
localTransport$.pipe(
switchMap(async (transport) => {
switchMap(async ({ transport }) => {
const transportSupportsDelegation =
"livekit_service_url" in transport &&
(await checkDelegationSupport(
@@ -5,7 +5,6 @@ SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
@@ -13,6 +12,7 @@ import { Config } from "../../../config/Config.ts";
import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
import { customTransport as customTransportSetting } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type TransportLocator } from "../../../livekit/auth/types.ts";
/**
* Determines the transport to advertise in our MatrixRTC membership and publish
@@ -20,7 +20,10 @@ import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
*/
export async function getLocalTransport(
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
): Promise<UnstableLivekitTransport> {
): Promise<TransportLocator> {
const serverName = client.getDomain();
if (serverName === null) throw new Error("No server name");
const discovery = new RtcTransportAutoDiscovery({
client,
resolvedConfig: Config.get(),
@@ -32,11 +35,10 @@ export async function getLocalTransport(
customTransportSetting.value$.value ??
(await discovery.discoverPreferredTransport());
if (transport === null)
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
if (transport === null) throw new MatrixRTCTransportMissingError(serverName);
// TODO: Since this module no longer maps auth errors to fatal user-facing
// errors (as it doesn't even perform auth), that needs to happen somewhere
// else.
return transport;
return { transport, serverName };
}
@@ -121,11 +121,12 @@ function setupTest(): void {
function setupRemoteConnection(): Connection {
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -169,11 +170,12 @@ describe("Start connection states", () => {
setupTest();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -188,11 +190,12 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -240,11 +243,12 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -301,11 +305,12 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -545,7 +550,8 @@ describe("remote track logging", () => {
roomId: ROOM_ID,
slotId: SLOT_ID,
role: "subscriber",
transport: transport,
transport,
serverName: "m.localhost",
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -51,6 +51,8 @@ export interface ConnectionOpts {
role: "publisher" | "subscriber";
/** The media transport to connect to. */
transport: UnstableLivekitTransport;
/** The name of the homeserver to which the {@link transport} belongs. */
serverName: string;
/** The Matrix client to use for OpenID and SFU config requests. */
client: ClientGetTokenParts & ClientOpenIDParts;
/** The ID of the Matrix room in which the session takes place. */
@@ -114,6 +116,11 @@ export class Connection {
*/
public readonly transport: UnstableLivekitTransport;
/**
* The name of the homeserver to which the {@link transport} belongs.
*/
public readonly serverName: string;
public readonly livekitRoom: LivekitRoom;
private scope: ObservableScope;
@@ -162,12 +169,13 @@ export class Connection {
`[Connection ${JSON.stringify(opts.transport)}]`,
);
this.logger.debug(`constructor called`);
const { role, transport, client, scope } = opts;
const { role, transport, serverName, client, scope } = opts;
this.scope = scope;
this.livekitRoom = opts.livekitRoomFactory();
this.role = role;
this.transport = transport;
this.serverName = serverName;
this.client = client;
this.remoteParticipants$ = scope.behavior(
@@ -401,6 +409,7 @@ export class Connection {
membership: this.ownMembershipIdentity,
role: this.role,
transport: this.transport,
serverName: this.serverName,
roomId: this.roomId,
slotId: this.slotId,
logger: this.logger,
@@ -46,6 +46,7 @@ export interface ConnectionFactory {
scope: ObservableScope,
role: "publisher" | "subscriber",
transport: UnstableLivekitTransport,
serverName: string,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
): Connection;
@@ -98,6 +99,7 @@ export class ECConnectionFactory implements ConnectionFactory {
*
* @param scope The observable scope (used for clean-up)
* @param transport The transport to use for this connection.
* @param serverName The name of the homeserver to which the {@link transport} belongs.
* @param ownMembershipIdentity required to connect (using the jwt service) with the SFU.
* @param logger The logger instance to use for this connection.
* @param sfuConfig optional config in case we already have a token for this connection.
@@ -107,6 +109,7 @@ export class ECConnectionFactory implements ConnectionFactory {
scope: ObservableScope,
role: "publisher" | "subscriber",
transport: UnstableLivekitTransport,
serverName: string,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
): Connection {
@@ -116,6 +119,7 @@ export class ECConnectionFactory implements ConnectionFactory {
slotId: this.slotId,
role,
transport,
serverName,
client: this.client,
scope: scope,
livekitRoomFactory: this.livekitRoomFactory,
@@ -15,23 +15,30 @@ import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
import {
createConnectionManager$,
ConnectionManagerData,
areUnstableLivekitTransportsEqual,
} from "./ConnectionManager.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type Behavior } from "../../Behavior.ts";
import { type TransportLocator } from "../../../livekit/auth";
// Some test constants
const TRANSPORT_1: UnstableLivekitTransport = {
const TRANSPORT_1: TransportLocator = {
transport: {
type: "livekit",
livekit_service_url: "https://lk.example.org",
},
serverName: "example.org",
};
const TRANSPORT_2: UnstableLivekitTransport = {
const TRANSPORT_2: TransportLocator = {
transport: {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
},
serverName: "sample.com",
};
let fakeConnectionFactory: ConnectionFactory;
@@ -94,13 +101,19 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn1 = connections.find((c) =>
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(
c.transport,
TRANSPORT_1.transport,
),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalled();
const conn2 = connections.find((c) =>
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(
c.transport,
TRANSPORT_2.transport,
),
);
expect(conn2).toBeDefined();
expect(conn2!.start).toHaveBeenCalled();
@@ -141,12 +154,18 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn2 = connections.find((c) =>
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(
c.transport,
TRANSPORT_2.transport,
),
);
expect(conn2).toBeDefined();
const conn1 = connections.find((c) =>
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(
c.transport,
TRANSPORT_1.transport,
),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalledOnce();
@@ -187,7 +206,10 @@ 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) =>
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(
c.transport,
TRANSPORT_2.transport,
),
);
expect(connection2).toBeDefined();
expect(connection2!.stop).toHaveBeenCalled();
@@ -279,7 +301,7 @@ describe("connectionManagerData$ stream", () => {
// Setup the fake participants streams behavior
// ==============================
fakeRemoteParticipantsStreams.set(
keyForTransport(TRANSPORT_1),
keyForTransport(TRANSPORT_1.transport),
behavior("oa-b", {
o: [],
a: [{ identity: "user1A" } as RemoteParticipant],
@@ -291,7 +313,7 @@ describe("connectionManagerData$ stream", () => {
);
fakeRemoteParticipantsStreams.set(
keyForTransport(TRANSPORT_2),
keyForTransport(TRANSPORT_2.transport),
behavior("o-a", {
o: [],
a: [{ identity: "user2A" } as RemoteParticipant],
@@ -20,8 +20,8 @@ import {
type ObservableScope,
} from "../../ObservableScope.ts";
import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type TransportLocator } from "../../../livekit/auth/types.ts";
export class ConnectionManagerData {
private readonly store: Map<
@@ -32,7 +32,7 @@ export class ConnectionManagerData {
public constructor(private readonly logger?: Logger) {}
public add(connection: Connection, participants: RemoteParticipant[]): void {
const key = this.getKey(connection.transport);
const key = this.getKey(connection);
const existing = this.store.get(key);
if (!existing) {
this.store.set(key, { connection, participants });
@@ -46,7 +46,7 @@ export class ConnectionManagerData {
}
}
private getKey(transport: UnstableLivekitTransport): string {
private getKey({ transport, serverName }: TransportLocator): 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 JSON.stringify([
@@ -54,6 +54,7 @@ export class ConnectionManagerData {
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined,
serverName,
]);
}
@@ -62,13 +63,13 @@ export class ConnectionManagerData {
}
public getConnectionForTransport(
transport: UnstableLivekitTransport,
transport: TransportLocator,
): Connection | null {
return this.store.get(this.getKey(transport))?.connection ?? null;
}
public getParticipantsForTransport(
transport: UnstableLivekitTransport,
transport: TransportLocator,
): RemoteParticipant[] {
const key = this.getKey(transport);
const existing = this.store.get(key);
@@ -79,11 +80,44 @@ export class ConnectionManagerData {
}
}
/**
* Unpacks an {@link UnstableLivekitTransport} into its constituent `url` and
* `serviceUrl`, discarding the type information that at least one of them must
* be defined.
*/
function unpackTransport(transport: UnstableLivekitTransport): {
url: string | undefined;
serviceUrl: string | undefined;
} {
return {
url: "url" in transport ? transport.url : undefined,
serviceUrl:
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined,
};
}
/**
* Constructs an {@link UnstableLivekitTransport} from its constituent `url` and
* `serviceUrl`. This assumes that at least one of the two are defined.
*/
function repackTransport(
url: string | undefined,
serviceUrl: string | undefined,
): UnstableLivekitTransport {
return {
type: "livekit",
...(url === undefined ? {} : { url }),
...(serviceUrl === undefined ? {} : { livekit_service_url: serviceUrl }),
} as UnstableLivekitTransport;
}
interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
localTransport: UnstableLivekitTransport;
remoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
localTransport: TransportLocator;
remoteTransports$: Behavior<Epoch<TransportLocator[]>>;
logger: Logger;
ownMembershipIdentity: CallMembershipIdentityParts;
@@ -129,11 +163,11 @@ export function createConnectionManager$({
* All transports currently managed by the ConnectionManager. This list does
* not include duplicate transports.
*/
const localAndRemoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>> =
const localAndRemoteTransports$: Behavior<Epoch<TransportLocator[]>> =
scope.behavior(
remoteTransports$.pipe(
mapEpoch((remoteTransports) =>
removeDuplicateTransports([localTransport, ...remoteTransports]),
removeDuplicateLocators([localTransport, ...remoteTransports]),
),
tap((transports) =>
logger.debug("localAndRemoteTransports$ = ", transports),
@@ -149,30 +183,32 @@ export function createConnectionManager$({
generateItemsWithEpoch(
"ConnectionManager connections$",
function* (transports) {
for (const transport of transports) {
for (const { transport, serverName } of transports) {
const role =
transport === localTransport
serverName === localTransport.serverName &&
areUnstableLivekitTransportsEqual(
transport,
localTransport.transport,
)
? ("publisher" as const)
: ("subscriber" as const);
const url = "url" in transport ? transport.url : undefined;
const serviceUrl =
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined;
yield { keys: [role, url, serviceUrl], data: undefined };
// In order to maintain a consistent set of map keys, we must
// temporarily downgrade `transport` to a weaker type
const { url, serviceUrl } = unpackTransport(transport);
yield {
keys: [role, url, serviceUrl, serverName],
data: undefined,
};
}
},
(scope, _data$, role, url, serviceUrl) => {
(scope, _data$, role, url, serviceUrl, serverName) => {
const connection = connectionFactory.createConnection(
scope,
role,
{
type: "livekit",
...(url === undefined ? {} : { url }),
...(serviceUrl === undefined
? {}
: { livekit_service_url: serviceUrl }),
} as UnstableLivekitTransport,
// Pack the transport keys back into a stronger
// `UnstableLivekitTransport` type
repackTransport(url, serviceUrl),
serverName,
ownMembershipIdentity,
logger,
);
@@ -230,12 +266,65 @@ export function createConnectionManager$({
return { connectionManagerData$ };
}
function removeDuplicateTransports<T extends UnstableLivekitTransport>(
transports: T[],
): T[] {
return transports.reduce((acc, transport) => {
if (!acc.some((t) => areUnstableLivekitTransportsEqual(t, transport)))
acc.push(transport);
return acc;
}, [] as T[]);
// TODO add this to the JS-SDK
export function areUnstableLivekitTransportsEqual<
T extends UnstableLivekitTransport,
>(t1: T | null, t2: T | null): boolean {
if (t1 && t2) {
if ("url" in t1 !== "url" in t2) return false;
if ("livekit_service_url" in t1 !== "livekit_service_url" in t2)
return false;
if (
"url" in t1 &&
t1.url !== (t2 as UnstableLivekitTransport & { url: string }).url
)
return false;
if (
"livekit_service_url" in t1 &&
t1.livekit_service_url !==
(t2 as UnstableLivekitTransport & { livekit_service_url: string })
.livekit_service_url
)
return false;
return true;
}
return !t1 && !t2;
}
function removeDuplicateLocators(
locators: TransportLocator[],
): TransportLocator[] {
// Set tracking which locators we have added to `deduped` so far
// url -> serviceUrl -> serverNames
const byUrl = new Map<
string | undefined,
Map<string | undefined, Set<string>>
>();
const deduped: TransportLocator[] = [];
for (const locator of locators) {
const { transport, serverName } = locator;
const url = "url" in transport ? transport.url : undefined;
const serviceUrl =
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined;
let byServiceUrl = byUrl.get(url);
if (byServiceUrl === undefined) {
byServiceUrl = new Map();
byUrl.set(url, byServiceUrl);
}
let serverNames = byServiceUrl.get(serviceUrl);
if (serverNames === undefined) {
serverNames = new Set();
byServiceUrl.set(serviceUrl, serverNames);
}
if (!serverNames.has(serverName)) {
serverNames.add(serverName);
deduped.push(locator);
}
}
return deduped;
}
@@ -97,6 +97,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
testScope,
"subscriber",
exampleTransport,
"example.org",
ownMemberMock,
logger,
);
@@ -147,6 +148,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
testScope,
"subscriber",
exampleTransport,
"example.org",
ownMemberMock,
logger,
);
@@ -184,6 +186,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
testScope,
"subscriber",
exampleTransport,
"example.org",
ownMemberMock,
logger,
);
@@ -223,6 +226,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
testScope,
"subscriber",
exampleTransport,
"example.org",
ownMemberMock,
logger,
);
@@ -263,6 +267,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
testScope,
"subscriber",
exampleTransport,
"example.org",
ownMemberMock,
logger,
);
@@ -18,12 +18,7 @@ import {
type RemoteMatrixLivekitMember,
createRemoteMatrixLivekitMembers$,
} from "./MatrixLivekitMembers.ts";
import {
Epoch,
mapEpoch,
ObservableScope,
trackEpoch,
} from "../../ObservableScope.ts";
import { Epoch, ObservableScope, trackEpoch } from "../../ObservableScope.ts";
import { ConnectionManagerData } from "./ConnectionManager.ts";
import {
flushPromises,
@@ -33,6 +28,8 @@ import {
import { type Connection } from "./Connection.ts";
import { constant } from "../../Behavior.ts";
import { localRtcMember } from "../../../utils/test-fixtures.ts";
import { type TransportLocator } from "../../../livekit/auth";
import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
let testScope: ObservableScope;
@@ -117,21 +114,15 @@ 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?: UnstableLivekitTransport }[]>
Epoch<
{ membership: CallMembership; transport: TransportLocator | undefined }[]
>
>;
} {
const memberships$ = m$.pipe(trackEpoch());
const membershipsWithTransport$ = memberships$.pipe(
mapEpoch((members) => {
return members.map((m) => {
const tr = m.getTransport();
return {
membership: m,
transport:
tr?.type === "livekit" ? (tr as UnstableLivekitTransport) : undefined,
};
});
}),
const memberships$ = testScope.behavior(m$.pipe(trackEpoch()));
const { membershipsWithTransport$ } = membershipsAndTransports$(
testScope,
memberships$,
);
return {
memberships$,
@@ -6,10 +6,7 @@ Please see LICENSE in the repository root for full details.
*/
import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
import {
type CallMembership,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, filter, map } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
@@ -18,6 +15,7 @@ import { type IConnectionManager } from "./ConnectionManager";
import { Epoch, type ObservableScope } from "../../ObservableScope";
import { type Connection } from "./Connection";
import { generateItemsWithEpoch } from "../../../utils/observable";
import { type TransportLocator } from "../../../livekit/auth";
interface LocalTaggedParticipant {
type: "local";
@@ -61,7 +59,7 @@ interface Props {
scope: ObservableScope;
membershipsWithTransport$: Behavior<
Epoch<
{ membership: CallMembership; transport?: UnstableLivekitTransport }[]
{ membership: CallMembership; transport: TransportLocator | undefined }[]
>
>;
connectionManager: IConnectionManager;
@@ -165,28 +163,3 @@ export function createRemoteMatrixLivekitMembers$({
}
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
// TODO add this to the JS-SDK
export function areUnstableLivekitTransportsEqual<
T extends UnstableLivekitTransport,
>(t1: T | null, t2: T | null): boolean {
if (t1 && t2) {
if ("url" in t1 !== "url" in t2) return false;
if ("livekit_service_url" in t1 !== "livekit_service_url" in t2)
return false;
if (
"url" in t1 &&
t1.url !== (t2 as UnstableLivekitTransport & { url: string }).url
)
return false;
if (
"livekit_service_url" in t1 &&
t1.livekit_service_url !==
(t2 as UnstableLivekitTransport & { livekit_service_url: string })
.livekit_service_url
)
return false;
return true;
}
return !t1 && !t2;
}
@@ -24,7 +24,7 @@ import {
type ClientOpenIDParts,
} from "../../../livekit/auth";
import {
exampleTransport,
exampleTransportLocator,
mockMediaDevices,
mockRtcMembership,
ownMemberMock,
@@ -32,11 +32,13 @@ import {
} from "../../../utils/test.ts";
import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
import {
areUnstableLivekitTransportsEqual,
createRemoteMatrixLivekitMembers$,
type RemoteMatrixLivekitMember,
} from "./MatrixLivekitMembers.ts";
import { createConnectionManager$ } from "./ConnectionManager.ts";
import {
areUnstableLivekitTransportsEqual,
createConnectionManager$,
} from "./ConnectionManager.ts";
import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
@@ -129,7 +131,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
const connectionManager = createConnectionManager$({
scope: testScope,
connectionFactory: ecConnectionFactory,
localTransport: exampleTransport,
localTransport: exampleTransportLocator,
remoteTransports$: membershipsAndTransports.transports$,
logger: logger,
ownMembershipIdentity: ownMemberMock,
+13 -9
View File
@@ -7,7 +7,6 @@ Please see LICENSE in the repository root for full details.
import {
type CallMembership,
type UnstableLivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
isUnstableLivekitTransport,
@@ -21,6 +20,7 @@ import {
type ObservableScope,
} from "./ObservableScope";
import { type Behavior } from "./Behavior";
import { type TransportLocator } from "../livekit/auth";
/**
* Tracks the transports used by ourselves, plus all other MatrixRTC session
@@ -31,16 +31,15 @@ export const membershipsAndTransports$ = (
memberships$: Behavior<Epoch<CallMembership[]>>,
): {
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
Epoch<
{ membership: CallMembership; transport: TransportLocator | undefined }[]
>
>;
transports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
transports$: Behavior<Epoch<TransportLocator[]>>;
} => {
const membershipsWithTransport$: Behavior<
Epoch<
{
membership: CallMembership;
transport: UnstableLivekitTransport | undefined;
}[]
{ membership: CallMembership; transport: TransportLocator | undefined }[]
>
> = scope.behavior(
memberships$.pipe(
@@ -49,14 +48,19 @@ export const membershipsAndTransports$ = (
const transport = membership.getTransport();
return {
membership,
transport: isUnstableLivekitTransport(transport) ? transport : undefined,
transport: isUnstableLivekitTransport(transport)
? {
transport,
serverName: membership.userId.replace(/^.*?:/, ""),
}
: undefined,
};
});
}),
),
);
const transports$: Behavior<Epoch<UnstableLivekitTransport[]>> = scope.behavior(
const transports$: Behavior<Epoch<TransportLocator[]>> = scope.behavior(
membershipsWithTransport$.pipe(
mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
),
+6
View File
@@ -78,6 +78,7 @@ import {
type RemoteScreenShareViewModel,
} from "../state/media/RemoteScreenShareViewModel";
import { Connection } from "../state/CallViewModel/remoteMembers/Connection";
import { type TransportLocator } from "../livekit/auth";
export function withFakeTimers(continuation: () => void): void {
vi.useFakeTimers();
@@ -213,6 +214,11 @@ export const exampleTransport: UnstableLivekitTransport = {
livekit_service_url: "https://lk.example.org",
};
export const exampleTransportLocator: TransportLocator = {
transport: exampleTransport,
serverName: "example.org",
};
export function mockRtcMembership(
user: string | RoomMember,
deviceId: string,