Compare commits

...
Author SHA1 Message Date
Timo K. 0dff10601f add full mesh tests 2026-10-26 23:17:27 +01:00
Timo K. 4b4881e22a fix widget test 2026-10-26 22:04:02 +01:00
Timo K. e7f4de1aa0 fmt 2026-10-26 22:04:02 +01:00
Timo K. 3141e3eb1e Update app.json 2026-10-26 22:04:02 +01:00
Timo K. 1686406585 Add full mesh draft MSC for reference 2026-10-26 22:04:02 +01:00
Timo K. b1baa7e77c use resolvedTransport to expose full_mesh dev tool data 2026-10-26 22:04:02 +01:00
Timo K. 89640bcc20 add full mesh backend 2026-10-26 22:04:02 +01:00
30 changed files with 4236 additions and 50 deletions
+3
View File
@@ -71,6 +71,9 @@
"device_id": "Device ID: {{id}}",
"duplicate_tiles_label": "Number of additional tile copies per participant",
"environment_variables": "Environment variables",
"full_mesh_peer_connections": "Full mesh peer connections",
"full_mesh_publishing": "Publish on full mesh (MSCXXXX) instead of LiveKit",
"full_mesh_transport": "Full mesh: {{device}}",
"hostname": "Hostname: {{hostname}}",
"livekit_server_info": "LiveKit Server Info",
"livekit_sfu": "LiveKit SFU: {{url}}",
+125
View File
@@ -0,0 +1,125 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { expect, type Page, test } from "@playwright/test";
import { widgetTest } from "../fixtures/widget-user";
import { HOST1, TestHelpers } from "./test-helpers";
/**
* Records the LiveKit signalling websockets a page opens: the SFU is reached
* with an access token in the url, which full mesh never needs.
*/
function recordSfuSockets(page: Page): string[] {
const urls: string[] = [];
page.on("websocket", (ws) => {
if (ws.url().includes("access_token")) urls.push(ws.url());
});
return urls;
}
const cases: { name: string; fullMesh: [boolean, boolean] }[] = [
// Both publish on full mesh: the media flows without any SFU
{ name: "both on full mesh", fullMesh: [true, true] },
// One on each: Florian receives Valere over LiveKit without publishing there
{ name: "full mesh and LiveKit mixed", fullMesh: [true, false] },
];
cases.forEach(({ name, fullMesh: [florianMesh, valereMesh] }) => {
widgetTest(`Full mesh call: ${name}`, async ({ addUser, browserName }) => {
test.skip(
browserName === "firefox",
"The is test is not working on firefox CI environment. No mic/audio device inputs so cam/mic are disabled",
);
test.slow();
// Distinct per case: the cases run in parallel and register at once
const suffix = `${Number(florianMesh)}${Number(valereMesh)}`;
const [florian, valere] = await Promise.all([
addUser(`florian${suffix}`, HOST1),
addUser(`valere${suffix}`, HOST1),
]);
// Element Web doesn't auto-approve the full mesh signalling for Element
// Call yet, so it asks for it whenever a widget with full mesh opens
for (const { page } of [florian, valere])
await page.addLocatorHandler(
page.getByRole("heading", { name: "Approve widget permissions" }),
async () => page.getByRole("button", { name: "Approve" }).click(),
);
const sockets = {
florian: recordSfuSockets(florian.page),
valere: recordSfuSockets(valere.page),
};
const roomName = "Full Mesh Room";
await TestHelpers.createRoom(roomName, florian.page, [valere.mxId]);
await TestHelpers.acceptRoomInvite(roomName, valere.page);
await TestHelpers.openWidgetSetEmbeddedElementCallRtcModeCloseWidget(
florian.page,
"2_0",
{ fullMesh: florianMesh },
);
await TestHelpers.openWidgetSetEmbeddedElementCallRtcModeCloseWidget(
valere.page,
"2_0",
{ fullMesh: valereMesh },
);
await TestHelpers.startCallInCurrentRoom(florian.page, false);
await TestHelpers.joinCallFromLobby(florian.page);
await TestHelpers.joinCallInCurrentRoom(valere.page);
// A widget only asks for the full mesh signalling with full mesh turned
// on, so a LiveKit publisher in Element Web can't receive full mesh yet:
// the other's media arrives if it is on LiveKit or we have full mesh
const receives = {
florian: !valereMesh || florianMesh,
valere: !florianMesh || valereMesh,
};
// In parallel: assertions on a page also run its permission handler
await Promise.all(
(["florian", "valere"] as const).map(async (name) => {
const frame = { florian, valere }[name].page
.locator('iframe[title="Element Call"]')
.contentFrame();
await expect(frame.getByTestId("videoTile")).toHaveCount(2, {
timeout: 15000,
});
const waiting = frame.getByText("Waiting for media...");
if (receives[name]) {
// Over whichever transport carries it
await expect(waiting).not.toBeVisible({ timeout: 20000 });
await TestHelpers.expectVisibleVideoCount(frame, 2);
} else {
await expect(waiting).toBeVisible({ timeout: 20000 });
await TestHelpers.expectVisibleVideoCount(frame, 1);
}
}),
);
if (florianMesh && valereMesh) {
// Nobody published on LiveKit, so nobody connected to an SFU
expect(sockets.florian).toEqual([]);
expect(sockets.valere).toEqual([]);
} else {
// Florian receives Valere's media from the SFU without publishing there
expect(sockets.florian.length).toBeGreaterThan(0);
}
// Florian sees Valere's camera go off: over the stream metadata for full
// mesh, without renegotiating
const valereFrame = valere.page
.locator('iframe[title="Element Call"]')
.contentFrame();
await valereFrame.getByRole("switch", { name: "Stop video" }).click();
const florianFrame = florian.page
.locator('iframe[title="Element Call"]')
.contentFrame();
await TestHelpers.expectVisibleVideoCount(florianFrame, 1);
});
});
+7 -1
View File
@@ -292,6 +292,7 @@ export class TestHelpers {
public static async openWidgetSetEmbeddedElementCallRtcModeCloseWidget(
page: Page,
mode: RtcMode,
{ fullMesh = false }: { fullMesh?: boolean } = {},
): Promise<void> {
await page.getByRole("button", { name: "Video call" }).click({
timeout: 5000,
@@ -300,7 +301,7 @@ export class TestHelpers {
timeout: 10000,
});
await TestHelpers.setEmbeddedElementCallRtcMode(page, mode);
await TestHelpers.setEmbeddedElementCallRtcMode(page, mode, { fullMesh });
await page.getByRole("button", { name: "Close lobby" }).click();
}
@@ -313,6 +314,7 @@ export class TestHelpers {
public static async setEmbeddedElementCallRtcMode(
page: Page,
mode: RtcMode,
{ fullMesh = false }: { fullMesh?: boolean } = {},
): Promise<void> {
const iframe = page.locator('iframe[title="Element Call"]').contentFrame();
@@ -330,6 +332,10 @@ export class TestHelpers {
// compat
await iframe.getByText("Compatibility: state events").click();
}
// Publish on the full mesh transport (MSCXXXX) instead of LiveKit
await iframe
.getByText("Publish on full mesh", { exact: false })
.setChecked(fullMesh);
await iframe.getByTestId("modal_close").click();
}
+9 -4
View File
@@ -66,9 +66,14 @@ media backend: whatever carries the media, behind the `MediaBackend` interface i
`sdk/src/media-backend/api.ts`. The client is the join of the two: a member exists
once its membership does, and gets its media once the backend has something
for that membership on its transport; `state$` is the membership state combined
with the local media connection. LiveKit is the one backend today; a
cascading SFU or full mesh is another folder under `sdk/src/media-backend/`, handed
in through `options.backend`, and nothing in the left box changes.
with the local media connection. Two backends exist, LiveKit (`livekit/`) and
full mesh (`full-mesh/`, MSCXXXX), and the client always runs both, combined
(`combineBackends.ts`). It publishes on the backend of the type of
`options.transport`, LiveKit without one, and hands every remote member to the
backend of that member's transport, so members on either are received. A
backend nobody is on connects nowhere. Another transport is another folder
under `sdk/src/media-backend/`, added to the combination, and nothing in the
left box changes; `options.backend` replaces the combination altogether.
### How Element Call uses it
@@ -149,7 +154,7 @@ the sections below still describe `MatrixRTCClient` until that lands.
| publish request | a `PublishRequest`: a source, where to capture it from and how to encode it. Given at the join in `publish`, or later to `publish` on the participation | |
| attach, detach | handing a `<video>` or `<audio>` element to a track and taking it back. The SDK owns the stream and the observers on the element | LiveKit `Track.attach` |
| transport | an `RTCTransport`: where media is exchanged, a LiveKit service url today. Carries the raw js-sdk `Transport` from the membership and, in `resolved$`, what the backend fetched to connect to it | MSC4143 "focus" |
| media backend | a `MediaBackend`: what carries the media, behind `sdk/src/media-backend/api.ts`. One per participation, chosen with `backend` in the join options; LiveKit under `sdk/src/media-backend/livekit/` is the one that exists | "SFU", a LiveKit room, a peer connection |
| media backend | a `MediaBackend`: what carries the media, behind `sdk/src/media-backend/api.ts`. One per participation: LiveKit and full mesh combined, publishing on the type of `transport` in the join options, unless `backend` replaces them | "SFU", a LiveKit room, a peer connection |
Names use `RTC` in capitals, as the js-sdk does: `RTCSlot`, `RTCParticipation`,
`RTCMember`, `matrixRTCMode`.
+8
View File
@@ -61,6 +61,14 @@ export {
createLivekitBackend,
type LivekitBackendOptions,
} from "./src/media-backend/livekit/LivekitBackend";
// Join with `transport: { type: FULL_MESH_TRANSPORT_TYPE }` to publish on full mesh
export {
createFullMeshBackend,
FULL_MESH_TRANSPORT_TYPE,
type FullMeshBackendOptions,
} from "./src/media-backend/full-mesh/FullMeshBackend";
// For a widget host to allow the full mesh signalling
export { signallingEventTypes as fullMeshSignallingEventTypes } from "./src/media-backend/full-mesh/signalling";
// For a host that creates LiveKit tracks itself, such as a preview before joining
export { convertToLivekitProcessor } from "./src/media-backend/livekit/videoProcessor";
// For a developer panel to check a transport before using it
+26 -3
View File
@@ -50,6 +50,11 @@ import {
type MediaBackendContext,
type MediaKey,
} from "./media-backend/api";
import { combineBackends } from "./media-backend/combineBackends";
import {
completeFullMeshTransport,
createFullMeshBackend,
} from "./media-backend/full-mesh/FullMeshBackend";
import { createLivekitBackend } from "./media-backend/livekit/LivekitBackend";
import { findRemotePublishLivekitTransport } from "./media-backend/livekit/RemotePublishTransport";
import { createHomeserverConnected$ } from "./matrixrtc/HomeserverConnected";
@@ -121,6 +126,13 @@ export function createRTCParticipation(
: `${userId}:${deviceId}`,
};
// The host names full mesh by its type alone; the device is ours
if (options.transport)
options = {
...options,
transport: completeFullMeshTransport(options.transport, deviceId),
};
const backend = createBackend(scope, client, options, {
roomId: room.roomId,
slotId: jsSdkSession.slotId,
@@ -310,15 +322,26 @@ export function createRTCParticipation(
};
}
/** The host's backend, or LiveKit configured from the participation options. */
/**
* The host's backend, or every backend the SDK has, combined: we publish on
* the type of the transport the host named (LiveKit without one) and receive
* from members on any of them.
*/
function createBackend(
scope: ObservableScope,
client: MatrixClient,
{ backend, mediaQuality }: RTCParticipationOptions,
{ backend, mediaQuality, transport }: RTCParticipationOptions,
context: MediaBackendContext,
): MediaBackend {
if (backend) return backend(scope, context);
return createLivekitBackend(scope, context, { client, mediaQuality });
return combineBackends(
scope,
[
createLivekitBackend(scope, context, { client, mediaQuality }),
createFullMeshBackend(scope, context, { client }),
],
transport?.type ?? "livekit",
);
}
/** The slot's per-participant keys: the ones it already holds on subscribe, then each new one. */
@@ -0,0 +1,91 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { describe, expect, it, vi } from "vitest";
import { BehaviorSubject, NEVER } from "rxjs";
import { type RTCMember, type TransportMetadata } from "../api";
import { type MemberMedia } from "../media-api";
import { constant } from "../reactive/Behavior";
import { ObservableScope } from "../reactive/ObservableScope";
import { type MediaBackend } from "./api";
import { combineBackends } from "./combineBackends";
function fakeBackend(type: string): MediaBackend & { ended: string[] } {
const ended: string[] = [];
return {
ended,
transportType: type,
prepareLocalTransport: vi.fn(async () =>
Promise.resolve({ canDelegateDelayedLeave: false }),
),
delegateDelayedLeave: vi.fn(),
local: {} as MediaBackend["local"],
mediaFor: (scope, member): MemberMedia => {
scope.onEnd(() => ended.push(member.memberId));
return {
tracks$: constant([{ id: `${type}:${member.memberId}` }] as never),
encryptionError$: NEVER,
};
},
connections$: constant([
{ transport: { type }, state: "connected" as never, resolved: undefined },
]),
setAudioOutputDeviceId: vi.fn(),
sendData: vi.fn(),
data$: NEVER,
};
}
function transport(type: string): TransportMetadata {
return { type, id: type, raw: { type }, resolved$: constant(undefined) };
}
describe("combineBackends", () => {
it("routes each member to the backend of its transport type", () => {
const scope = new ObservableScope();
const livekit = fakeBackend("livekit");
const mesh = fakeBackend("full_mesh");
const combined = combineBackends(scope, [livekit, mesh], "livekit");
expect(combined.transportType).toBe("livekit");
expect(combined.connections$.value).toHaveLength(2);
const transport$ = new BehaviorSubject<TransportMetadata | undefined>(
transport("full_mesh"),
);
const memberScope = new ObservableScope();
const media = combined.mediaFor(memberScope, {
memberId: "m1",
transport$,
} as unknown as RTCMember);
expect(media.tracks$.value).toEqual([{ id: "full_mesh:m1" }]);
transport$.next(transport("livekit"));
expect(mesh.ended).toEqual(["m1"]);
expect(media.tracks$.value).toEqual([{ id: "livekit:m1" }]);
transport$.next(transport("unknown"));
expect(media.tracks$.value).toBeNull();
// The publishing backend keeps every member for as long as it is one
expect(livekit.ended).toEqual([]);
memberScope.end();
expect(livekit.ended).toEqual(["m1"]);
scope.end();
});
it("publishes on the backend of the publish type", () => {
const scope = new ObservableScope();
const livekit = fakeBackend("livekit");
const mesh = fakeBackend("full_mesh");
const combined = combineBackends(scope, [livekit, mesh], "full_mesh");
expect(combined.transportType).toBe("full_mesh");
expect(() => combineBackends(scope, [livekit, mesh], "other")).toThrow(
"No media backend for transport other",
);
scope.end();
});
});
+80
View File
@@ -0,0 +1,80 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { combineLatest, map, merge, NEVER, of, switchMap } from "rxjs";
import { type RTCMember } from "../api";
import { MatrixRTCError } from "../errors";
import { type MemberMedia } from "../media-api";
import { type ObservableScope } from "../reactive/ObservableScope";
import { mapScoped } from "../utils/mapScoped";
import { type MediaBackend } from "./api";
/**
* Several backends as one: we publish on the backend of `publishType`, and
* receive from every member on a transport any of the backends serves. A
* member is handed to the backend of its current transport type. A backend
* nobody is on costs nothing: it connects only for the members handed to it.
*/
export function combineBackends(
scope: ObservableScope,
all: MediaBackend[],
publishType: string,
): MediaBackend {
const byType = new Map(all.map((b) => [b.transportType, b]));
const publish = byType.get(publishType);
if (publish === undefined)
throw new MatrixRTCError(`No media backend for transport ${publishType}`);
const mediaFor = (
memberScope: ObservableScope,
member: RTCMember,
): MemberMedia => {
// The publishing backend sees every member, whatever its transport, so
// that it can tell members from strangers asking for our media (full
// mesh subscribers on other transports). It connects nowhere for them
const publishMedia = publish.mediaFor(memberScope, member);
const type$ = memberScope.behavior(
member.transport$.pipe(map((transport) => transport?.type)),
);
const media$ = mapScoped(memberScope, type$, (s, type) =>
type === publish.transportType
? publishMedia
: byType.get(type)?.mediaFor(s, member),
);
return {
tracks$: memberScope.behavior(
media$.pipe(switchMap((media) => media?.tracks$ ?? of(null))),
),
encryptionError$: media$.pipe(
switchMap((media) => media?.encryptionError$ ?? NEVER),
),
};
};
return {
transportType: publish.transportType,
prepareLocalTransport: async (transport) =>
publish.prepareLocalTransport(transport),
delegateDelayedLeave: async (delayId) =>
publish.delegateDelayedLeave(delayId),
local: publish.local,
mediaFor,
connections$: scope.behavior(
combineLatest(all.map((b) => b.connections$)).pipe(
map((lists) => lists.flat()),
),
),
setAudioOutputDeviceId: async (deviceId) => {
await Promise.all(
all.map(async (b) => b.setAudioOutputDeviceId(deviceId)),
);
},
sendData: async (topic, text) => publish.sendData(topic, text),
data$: merge(...all.map((b) => b.data$)),
};
}
@@ -0,0 +1,432 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { EventEmitter } from "events";
import {
ClientEvent,
EventType,
type ITurnServer,
MatrixEvent,
} from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, NEVER } from "rxjs";
import { type RTCMember, type TransportMetadata } from "../../api";
import { defaultSessionTimings } from "../../config";
import { E2eeType } from "../../encryption";
import { constant } from "../../reactive/Behavior";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type MediaBackend, MediaConnectionState } from "../api";
import {
completeFullMeshTransport,
createFullMeshBackend,
type FullMeshBackendOptions,
} from "./FullMeshBackend";
import { FULL_MESH_TRANSPORT_TYPE, SignallingEventType } from "./signalling";
import {
FakePeerConnection,
installWebRtcFakes,
lastPeerConnection,
peer,
settle,
} from "./test";
const roomId = "!room:example.org";
const slotId = "m.call#ROOM";
const own = { userId: "@me:example.org", deviceId: "MYDEVICE", memberId: "me" };
const fullMesh = (deviceId: string): Transport => ({
type: FULL_MESH_TRANSPORT_TYPE,
device_id: deviceId,
});
const livekit: Transport = { type: "livekit", livekit_service_url: "x" };
function metadata(raw: Transport): TransportMetadata {
return {
type: raw.type,
id: JSON.stringify(raw),
raw,
resolved$: constant(undefined),
};
}
describe("createFullMeshBackend", () => {
let scope: ObservableScope;
let client: EventEmitter & FullMeshBackendOptions["client"];
let send: ReturnType<typeof vi.fn>;
let turnServers: ITurnServer[];
let backend: MediaBackend;
const seq = new Map<string, number>();
/** A to-device message from `peer` (or another sender), as the sync would deliver it. */
function deliver(
type: string,
connectionId: string,
content: Record<string, unknown> = {},
from = peer,
): void {
const next = seq.get(from.memberId) ?? 0;
seq.set(from.memberId, next + 1);
const event = new MatrixEvent({
type,
sender: from.userId,
content: {
room_id: roomId,
slot_id: slotId,
member_id: from.memberId,
target_member_id: own.memberId,
device_id: from.deviceId,
connection_id: connectionId,
seq: next,
...content,
},
});
vi.spyOn(event, "getWireType").mockReturnValue(
EventType.RoomMessageEncrypted,
);
client.emit(ClientEvent.ToDeviceEvent, event);
}
/** What we sent, as [type, recipient device, content]. */
function sent(type?: string): [string, string, Record<string, unknown>][] {
return send.mock.calls
.filter(([t]) => type === undefined || t === type)
.map(([t, [target], content]) => [t, target.deviceId, content]);
}
/** A member of the slot, as the session hands it to the backend. */
function addMember(
transport: Transport,
who = peer,
): { scope: ObservableScope; tracks: () => unknown } {
const memberScope = new ObservableScope();
const media = backend.mediaFor(memberScope, {
...who,
transport$: new BehaviorSubject(metadata(transport)),
} as unknown as RTCMember);
return { scope: memberScope, tracks: () => media.tracks$.value };
}
beforeEach(() => {
vi.useFakeTimers();
installWebRtcFakes();
seq.clear();
scope = new ObservableScope();
send = vi.fn(async () => Promise.resolve());
turnServers = [];
client = Object.assign(new EventEmitter(), {
encryptAndSendToDevice: send,
getTurnServers: (): ITurnServer[] => turnServers,
checkTurnServers: vi.fn(async () => {
turnServers = [{ urls: ["turn:x"], username: "u", credential: "p" }];
return Promise.resolve(true);
}),
}) as unknown as EventEmitter & FullMeshBackendOptions["client"];
backend = createFullMeshBackend(
scope,
{
roomId,
slotId,
ownMembershipIdentity: own,
publish: [],
encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
mediaKeys$: NEVER,
timings: defaultSessionTimings,
logger,
},
{ client },
);
});
afterEach(() => {
scope.end();
vi.useRealTimers();
vi.unstubAllGlobals();
});
it("only prepares a full mesh transport", async () => {
await expect(backend.prepareLocalTransport(livekit)).rejects.toThrow(
"Not a full mesh transport: livekit",
);
await expect(
backend.prepareLocalTransport(fullMesh(own.deviceId)),
).resolves.toEqual({ canDelegateDelayedLeave: false });
expect(backend.local.connectionState$.value).toBe(
MediaConnectionState.Connected,
);
});
it("fetches TURN servers when the client has none yet", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
expect(client.checkTurnServers).toHaveBeenCalledOnce();
addMember(fullMesh(peer.deviceId));
await settle();
expect(lastPeerConnection().config.iceServers).toEqual([
{ urls: ["turn:x"], username: "u", credential: "p" },
]);
});
describe("as a publisher", () => {
it("turns subscribers away while it doesn't publish", async () => {
addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
expect(sent()).toEqual([
[
SignallingEventType.Close,
peer.deviceId,
expect.objectContaining({
reason: "not_publishing",
connection_id: "c1",
}),
],
]);
});
it("offers to a member on any transport", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
expect(sent(SignallingEventType.Description)).toEqual([
[
SignallingEventType.Description,
peer.deviceId,
expect.objectContaining({
connection_id: "c1",
target_member_id: peer.memberId,
description: { type: "offer", sdp: "offer-1" },
}),
],
]);
});
it("waits for the membership of a subscriber that is ahead of it", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
deliver(SignallingEventType.Subscribe, "c1");
await vi.advanceTimersByTimeAsync(5_000);
expect(sent()).toEqual([]);
addMember(livekit);
await settle();
expect(sent(SignallingEventType.Description)).toHaveLength(1);
});
it("rejects a subscriber that never becomes a member", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
deliver(SignallingEventType.Subscribe, "c1");
await vi.advanceTimersByTimeAsync(10_000);
expect(sent()).toEqual([
[
SignallingEventType.Close,
peer.deviceId,
expect.objectContaining({ reason: "not_a_member" }),
],
]);
expect(FakePeerConnection.instances).toEqual([]);
});
it("rejects a subscriber claiming another device of the member", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
addMember(livekit);
deliver(
SignallingEventType.Subscribe,
"c1",
{},
{
...peer,
deviceId: "OTHERDEVICE",
},
);
await settle();
expect(sent()).toEqual([
[
SignallingEventType.Close,
"OTHERDEVICE",
expect.objectContaining({ reason: "not_a_member" }),
],
]);
});
it("only serves the newest of several waiting subscribe requests", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
deliver(SignallingEventType.Subscribe, "c1");
deliver(SignallingEventType.Subscribe, "c2");
addMember(livekit);
await settle();
expect(
sent(SignallingEventType.Description).map(([, , c]) => c.connection_id),
).toEqual(["c2"]);
});
it("replaces the connection of a member that subscribes again", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
const first = lastPeerConnection();
deliver(SignallingEventType.Subscribe, "c2");
await settle();
expect(first.closed).toBe(true);
expect(lastPeerConnection()).not.toBe(first);
});
it("hands the subscriber's messages to its connection", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
deliver(SignallingEventType.Description, "c1", {
description: { type: "answer", sdp: "answer" },
});
deliver(SignallingEventType.Candidates, "c1", {
candidates: [{ candidate: "x" }],
});
// A late message of an earlier connection goes nowhere
deliver(SignallingEventType.Candidates, "c0", {
candidates: [{ candidate: "stale" }],
});
await settle();
const pc = lastPeerConnection();
expect(pc.remoteDescription).toEqual({ type: "answer", sdp: "answer" });
expect(pc.addedCandidates).toEqual([{ candidate: "x" }]);
});
it("stops sending to a member that leaves", async () => {
await backend.prepareLocalTransport(fullMesh(own.deviceId));
const member = addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
member.scope.end();
expect(lastPeerConnection().closed).toBe(true);
expect(sent(SignallingEventType.Close)).toEqual([
[
SignallingEventType.Close,
peer.deviceId,
expect.objectContaining({ connection_id: "c1" }),
],
]);
});
it("lists its subscribers for the developer tools", async () => {
expect(backend.connections$.value).toEqual([]);
await backend.prepareLocalTransport(fullMesh(own.deviceId));
expect(backend.connections$.value).toEqual([
{
transport: fullMesh(own.deviceId),
state: MediaConnectionState.Connected,
resolved: {
type: FULL_MESH_TRANSPORT_TYPE,
role: "publisher",
peerConnections: [],
},
},
]);
addMember(livekit);
deliver(SignallingEventType.Subscribe, "c1");
await settle();
expect(backend.connections$.value[0].resolved).toMatchObject({
peerConnections: [
{
memberId: peer.memberId,
connectionId: "c1",
connectionState: "new",
},
],
});
});
});
describe("as a subscriber", () => {
it("subscribes to a member publishing on full mesh", async () => {
const member = addMember(fullMesh(peer.deviceId));
await settle();
expect(sent()).toEqual([
[
SignallingEventType.Subscribe,
peer.deviceId,
expect.objectContaining({
room_id: roomId,
slot_id: slotId,
member_id: own.memberId,
device_id: own.deviceId,
target_member_id: peer.memberId,
seq: 0,
}),
],
]);
expect(member.tracks()).toBeNull();
expect(backend.connections$.value).toEqual([
expect.objectContaining({
transport: fullMesh(peer.deviceId),
resolved: expect.objectContaining({ role: "subscriber" }),
}),
]);
});
it("ignores a full mesh transport advertising another device", async () => {
addMember(fullMesh("OTHERDEVICE"));
await settle();
expect(sent()).toEqual([]);
});
it("leaves members on other transports alone", async () => {
const member = addMember(livekit);
await settle();
expect(sent()).toEqual([]);
expect(member.tracks()).toBeNull();
});
it("hands the publisher's messages to the subscription", async () => {
addMember(fullMesh(peer.deviceId));
await settle();
const [[, , subscribe]] = sent();
lastPeerConnection().receive("video", "user");
deliver(
SignallingEventType.Description,
subscribe.connection_id as string,
{
description: { type: "offer", sdp: "offer" },
stream_metadata: { user: { purpose: "m.usermedia" } },
},
);
await settle();
expect(sent(SignallingEventType.Description)).toEqual([
[
SignallingEventType.Description,
peer.deviceId,
expect.objectContaining({
description: { type: "answer", sdp: "answer" },
}),
],
]);
});
it("closes the subscription when the member leaves", async () => {
const member = addMember(fullMesh(peer.deviceId));
await settle();
member.scope.end();
expect(lastPeerConnection().closed).toBe(true);
expect(backend.connections$.value).toEqual([]);
});
});
it("has no data channel and doesn't take over the delayed leave", async () => {
await expect(backend.sendData("topic", "text")).rejects.toThrow();
await expect(backend.delegateDelayedLeave("id")).rejects.toThrow();
});
});
describe("completeFullMeshTransport", () => {
it("adds our device to a full mesh transport only", () => {
expect(
completeFullMeshTransport({ type: FULL_MESH_TRANSPORT_TYPE }, "DEV"),
).toEqual({ type: FULL_MESH_TRANSPORT_TYPE, device_id: "DEV" });
expect(completeFullMeshTransport(livekit, "DEV")).toBe(livekit);
});
});
@@ -0,0 +1,393 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
/**
* The full mesh transport (MSCXXXX): one WebRTC connection per subscriber
* and publisher, signalled over to-device messages, through the TURN server
* of each side's homeserver. Works subscribe-only too: without
* `prepareLocalTransport` it publishes nothing and still receives from every
* member on full mesh.
*/
import { type MatrixClient } from "matrix-js-sdk";
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
import {
BehaviorSubject,
combineLatest,
distinctUntilChanged,
filter,
firstValueFrom,
map,
NEVER,
type Observable,
of,
Subject,
switchMap,
timeout,
} from "rxjs";
import { type RTCMember } from "../../api";
import { MatrixRTCError } from "../../errors";
import { type MemberMedia } from "../../media-api";
import { type Behavior } from "../../reactive/Behavior";
import { ObservableScope } from "../../reactive/ObservableScope";
import { mapScoped } from "../../utils/mapScoped";
import {
type BackendConnection,
type MediaBackend,
type MediaBackendContext,
MediaConnectionState,
} from "../api";
import { createFullMeshLocalMedia } from "./LocalMedia";
import { type PeerConnectionBase } from "./PeerConnection";
import {
createPublisherConnection,
type PublisherConnection,
} from "./PublisherConnection";
import {
createSignalling,
FULL_MESH_TRANSPORT_TYPE,
type Peer,
type SignallingClient,
SignallingEventType,
type SignallingMessage,
} from "./signalling";
import {
createSubscriberConnection,
type SubscriberConnection,
} from "./SubscriberConnection";
export { FULL_MESH_TRANSPORT_TYPE } from "./signalling";
/** How long a subscribe request waits for the subscriber's membership to arrive. */
const SUBSCRIBE_GRACE_MS = 10_000;
export interface FullMeshBackendOptions {
client: SignallingClient &
Pick<MatrixClient, "getTurnServers" | "checkTurnServers">;
}
/**
* The transport object a member publishing on full mesh advertises: a host
* passes `{ type: FULL_MESH_TRANSPORT_TYPE }`, and the device is filled in.
* Any other transport is returned as it is.
*/
export function completeFullMeshTransport(
transport: Transport,
deviceId: string,
): Transport {
return transport.type === FULL_MESH_TRANSPORT_TYPE
? { ...transport, device_id: deviceId }
: transport;
}
export function createFullMeshBackend(
scope: ObservableScope,
{
roomId,
slotId,
ownMembershipIdentity,
publish,
logger: parentLogger,
}: MediaBackendContext,
{ client }: FullMeshBackendOptions,
): MediaBackend {
const logger = parentLogger.getChild("[FullMeshBackend]");
const signalling = createSignalling({
scope,
client,
roomId,
slotId,
own: ownMembershipIdentity,
logger,
});
const iceServers = async (): Promise<RTCIceServer[]> => {
if (client.getTurnServers().length === 0)
await client
.checkTurnServers()
.catch((e: unknown) => logger.warn("No TURN servers", e));
return client.getTurnServers().map(({ urls, username, credential }) => ({
urls,
username,
credential,
}));
};
let localTransport: Transport | null = null;
const prepared$ = new BehaviorSubject(false);
const audioOutputDeviceId$ = new BehaviorSubject<string | undefined>(
undefined,
);
const members = new Map<string, RTCMember>();
const memberAdded$ = new Subject<RTCMember>();
const subscribers = new Map<
string,
{ transport: Transport; peer: Peer; connection: SubscriberConnection }
>();
const publishers = new Map<
string,
{ connection: PublisherConnection; peer: Peer; scope: ObservableScope }
>();
scope.onEnd(() => {
for (const { scope: s } of Array.from(publishers.values())) s.end();
});
/** The connection id of the newest subscribe request per member. */
const latestSubscribe = new Map<string, string>();
// For the developer tools: one entry per publisher we subscribe to, and one
// for our own transport with every subscriber we send to. `resolved` lists
// the peer connections, as an SFU's entry lists what it connected to
const connectionsChanged$ = new BehaviorSubject<void>(undefined);
const connections$ = scope.behavior<BackendConnection[]>(
combineLatest([connectionsChanged$, prepared$]).pipe(
switchMap(([, prepared]) => {
const entries: Observable<BackendConnection>[] = [];
for (const { transport, peer, connection } of subscribers.values()) {
const base = connection.current$.value;
if (base)
entries.push(
base.state$.pipe(
map((state) => ({
transport,
state,
resolved: {
type: FULL_MESH_TRANSPORT_TYPE,
role: "subscriber",
peerConnections: [describe(peer, base)],
},
})),
),
);
}
if (localTransport && prepared) {
const transport = localTransport;
const outgoing = [...publishers.values()];
entries.push(
(outgoing.length === 0
? of([])
: combineLatest(outgoing.map((p) => p.connection.state$))
).pipe(
map(() => ({
transport,
// Publishing needs no server, so it is up once prepared
state: MediaConnectionState.Connected,
resolved: {
type: FULL_MESH_TRANSPORT_TYPE,
role: "publisher",
peerConnections: outgoing.map(({ peer, connection }) =>
describe(peer, connection),
),
},
})),
),
);
}
return entries.length === 0 ? of([]) : combineLatest(entries);
}),
),
);
const local = createFullMeshLocalMedia({
scope,
publish,
connectionState$: scope.behavior(
prepared$.pipe(
map((prepared) =>
prepared
? MediaConnectionState.Connected
: MediaConnectionState.Initialized,
),
),
),
peerConnections: () =>
[...publishers.values()].map(({ connection }) => connection.pc),
logger,
});
const acceptSubscribe = async (message: SignallingMessage): Promise<void> => {
const peer: Peer = {
userId: message.userId,
deviceId: message.deviceId,
memberId: message.memberId,
};
// We don't advertise full mesh, so nobody should have asked
if (!prepared$.value) {
signalling.send(peer, message.connectionId, {
type: SignallingEventType.Close,
reason: "not_publishing",
});
return;
}
latestSubscribe.set(peer.memberId, message.connectionId);
// The request may overtake the subscriber's membership
const member =
members.get(peer.memberId) ??
(await firstValueFrom(
memberAdded$.pipe(
filter((m) => m.memberId === peer.memberId),
timeout(SUBSCRIBE_GRACE_MS),
),
).catch(() => undefined));
if (latestSubscribe.get(peer.memberId) !== message.connectionId) return;
if (
member === undefined ||
member.userId !== peer.userId ||
member.deviceId !== peer.deviceId
) {
logger.warn(`Rejecting a subscribe from ${peer.memberId}`);
signalling.send(peer, message.connectionId, {
type: SignallingEventType.Close,
reason: "not_a_member",
});
return;
}
// At most one connection per member pair: the new one replaces the old
publishers.get(peer.memberId)?.scope.end();
const pubScope = new ObservableScope();
const connection = createPublisherConnection({
scope: pubScope,
peer,
connectionId: message.connectionId,
signalling,
iceServers: await iceServers(),
local,
logger,
});
publishers.set(peer.memberId, { connection, peer, scope: pubScope });
pubScope.onEnd(() => {
if (publishers.get(peer.memberId)?.connection === connection) {
publishers.delete(peer.memberId);
connectionsChanged$.next();
}
});
connectionsChanged$.next();
};
signalling.messages$.pipe(scope.bind()).subscribe((message) => {
const subscriber = subscribers.get(message.memberId)?.connection;
if (subscriber?.current$.value?.connectionId === message.connectionId)
return subscriber.handle(message);
const publisher = publishers.get(message.memberId)?.connection;
if (publisher?.connectionId === message.connectionId)
return publisher.handle(message);
if (message.type === SignallingEventType.Subscribe)
return void acceptSubscribe(message);
logger.debug(`Dropping ${message.type} of a stale connection`);
});
const mediaFor = (
memberScope: ObservableScope,
member: RTCMember,
): MemberMedia => {
members.set(member.memberId, member);
memberAdded$.next(member);
memberScope.onEnd(() => {
members.delete(member.memberId);
// We stop sending to members that are no longer joined
publishers.get(member.memberId)?.scope.end();
});
// In encrypted rooms the advertised device has to be the one the
// membership came from
const fullMesh$ = memberScope.behavior(
member.transport$.pipe(
map((transport) =>
transport?.type === FULL_MESH_TRANSPORT_TYPE &&
transport.raw.device_id === member.deviceId
? transport.raw
: undefined,
),
distinctUntilChanged((a, b) => a?.device_id === b?.device_id),
),
);
const subscriber$ = mapScoped(memberScope, fullMesh$, (s, transport) => {
const connection = createSubscriberConnection({
scope: s,
peer: {
userId: member.userId,
deviceId: member.deviceId,
memberId: member.memberId,
},
signalling,
iceServers,
audioOutputDeviceId$,
logger,
});
subscribers.set(member.memberId, {
transport,
peer: {
userId: member.userId,
deviceId: member.deviceId,
memberId: member.memberId,
},
connection,
});
connection.current$
.pipe(s.bind())
.subscribe(() => connectionsChanged$.next());
s.onEnd(() => {
subscribers.delete(member.memberId);
connectionsChanged$.next();
});
return connection;
});
return {
tracks$: memberScope.behavior(
subscriber$.pipe(switchMap((s) => s?.tracks$ ?? of(null))),
),
encryptionError$: NEVER,
};
};
return {
transportType: FULL_MESH_TRANSPORT_TYPE,
prepareLocalTransport: async (transport) => {
if (transport.type !== FULL_MESH_TRANSPORT_TYPE)
throw new MatrixRTCError(
`Not a full mesh transport: ${transport.type}`,
);
localTransport = transport;
// Fetched now, so that the first subscriber isn't kept waiting
await iceServers();
prepared$.next(true);
return { canDelegateDelayedLeave: false };
},
delegateDelayedLeave: async () =>
Promise.reject(
new MatrixRTCError("Full mesh cannot take over the delayed leave"),
),
local,
mediaFor,
connections$: connections$ as Behavior<BackendConnection[]>,
setAudioOutputDeviceId: async (deviceId) => {
audioOutputDeviceId$.next(deviceId);
return Promise.resolve();
},
sendData: async () =>
Promise.reject(new MatrixRTCError("Full mesh has no data channel yet")),
data$: NEVER,
};
}
/** One peer connection as the developer tools show it. */
function describe(
{ userId, deviceId, memberId }: Peer,
{ connectionId, pc }: PeerConnectionBase,
): Record<string, string> {
return {
userId,
deviceId,
memberId,
connectionId,
connectionState: pc.connectionState,
iceConnectionState: pc.iceConnectionState,
signalingState: pc.signalingState,
};
}
@@ -0,0 +1,456 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
BehaviorSubject,
combineLatest,
filter,
firstValueFrom,
map,
NEVER,
of,
switchMap,
} from "rxjs";
import { v4 as uuidv4 } from "uuid";
import { MatrixRTCError, toMatrixRTCError } from "../../errors";
import {
type LocalAudioMediaTrack,
type LocalVideoMediaTrack,
type MediaSource,
type PublishRequest,
type VideoProcessor,
} from "../../media-api";
import { type Behavior } from "../../reactive/Behavior";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type LocalMediaBackend, type MediaConnectionState } from "../api";
import { createFullMeshMediaTrack, rtpStats } from "./MediaTrack";
import { type StreamMetadata } from "./signalling";
type PublishedSource = Exclude<MediaSource, "unknown">;
/** One local track as the publisher connections send it. */
export interface LocalPublication {
source: PublishedSource;
kind: "audio" | "video";
/**
* The `MediaStream` it is sent in: one for the user media, one for the
* screen share. Its id is the `msid` the stream metadata is keyed by.
*/
stream: MediaStream;
/** What goes on the wire; null while the source is off. */
track$: Behavior<MediaStreamTrack | null>;
mediaTrack: LocalAudioMediaTrack | LocalVideoMediaTrack;
}
export interface FullMeshLocalMedia extends LocalMediaBackend {
/** Every local track, once the first `setPublishing(true)` started the capture. */
publications$: Behavior<LocalPublication[]>;
/** Whether the tracks reach the publisher connections. */
publishing$: Behavior<boolean>;
streamMetadata$: Behavior<StreamMetadata>;
}
interface Props {
scope: ObservableScope;
publish: PublishRequest[];
connectionState$: Behavior<MediaConnectionState | Error>;
/** The publisher connections, for the outbound statistics of a track. */
peerConnections: () => RTCPeerConnection[];
logger: Logger;
}
interface Capture {
/** What was captured, before any processor. */
raw: MediaStreamTrack;
out: MediaStreamTrack;
}
/** The local member's side: capture, the publication list and the controls. */
export function createFullMeshLocalMedia({
scope,
publish: initialRequests,
connectionState$,
peerConnections,
logger: parentLogger,
}: Props): FullMeshLocalMedia {
const logger = parentLogger.getChild("[LocalMedia]");
// Only their ids matter: the tracks are added per connection
const userMediaStream = new MediaStream();
const screenShareStream = new MediaStream();
const desired = new Map<PublishRequest["source"], PublishRequest>(
initialRequests.map((r) => [r.source, r]),
);
const started$ = new BehaviorSubject(false);
const publishing$ = new BehaviorSubject(false);
const publishError$ = new BehaviorSubject<Error | null>(null);
const publications$ = new BehaviorSubject<LocalPublication[]>([]);
const teardowns = new Map<PublishedSource, () => void>();
const statsFor = async (track: MediaStreamTrack | null) => {
for (const pc of peerConnections())
if (pc.getSenders().some((s) => s.track === track && track !== null))
return rtpStats(pc, track, "outbound-rtp");
return undefined;
};
/** Adds a publication whose track is managed by `control`. */
const addPublication = (
source: PublishedSource,
kind: "audio" | "video",
stream: MediaStream,
initial: MediaStreamTrack,
control: {
setEnabled: (
enabled: boolean,
track$: BehaviorSubject<MediaStreamTrack | null>,
) => Promise<boolean>;
setDevice?: (
deviceId: string,
track$: BehaviorSubject<MediaStreamTrack | null>,
) => Promise<void>;
switchFacingMode?: (
track$: BehaviorSubject<MediaStreamTrack | null>,
) => Promise<string | undefined>;
setProcessor?: (
p: VideoProcessor | undefined,
track$: BehaviorSubject<MediaStreamTrack | null>,
) => Promise<void>;
stop: () => void;
},
): LocalPublication => {
const pubScope = new ObservableScope();
const track$ = new BehaviorSubject<MediaStreamTrack | null>(initial);
const enabled$ = new BehaviorSubject(true);
const base = createFullMeshMediaTrack(pubScope, {
source,
kind,
id: uuidv4(),
track$,
muted$: pubScope.behavior(enabled$.pipe(map((e) => !e))),
stats: async () => statsFor(track$.value),
});
const local = {
setEnabled: async (enabled: boolean): Promise<boolean> => {
const result = await control.setEnabled(enabled, track$);
enabled$.next(result);
return result;
},
setDevice: async (deviceId: string): Promise<void> => {
if (!control.setDevice)
throw new MatrixRTCError(`A ${source} track has no device`);
await control.setDevice(deviceId, track$);
},
};
const mediaTrack =
base.kind === "audio"
? { ...base, ...local }
: {
...base,
...local,
facingMode$: pubScope.behavior(
track$.pipe(map((t) => facingModeOf(t))),
),
switchFacingMode: async () => control.switchFacingMode?.(track$),
setProcessor: async (p: VideoProcessor | undefined) =>
control.setProcessor?.(p, track$),
};
const publication: LocalPublication = {
source,
kind,
stream,
track$,
mediaTrack: mediaTrack as LocalAudioMediaTrack | LocalVideoMediaTrack,
};
publications$.next([...publications$.value, publication]);
teardowns.set(source, () => {
control.stop();
track$.value?.stop();
pubScope.end();
publications$.next(publications$.value.filter((p) => p !== publication));
teardowns.delete(source);
});
return publication;
};
const captureMicrophone = async (
request: Extract<PublishRequest, { source: "microphone" }>,
): Promise<void> => {
const track = await getTrack({
audio: { deviceId: exact(request.deviceId), ...request.capture },
});
addPublication("microphone", "audio", userMediaStream, track, {
// Muting a microphone keeps the device: unmuting is instant
setEnabled: async (enabled, track$) => {
if (track$.value) track$.value.enabled = enabled;
return Promise.resolve(enabled);
},
setDevice: async (deviceId, track$) => {
const next = await getTrack({
audio: { deviceId: { exact: deviceId }, ...request.capture },
});
next.enabled = track$.value?.enabled ?? true;
track$.value?.stop();
track$.next(next);
},
stop: () => {},
});
};
const captureCamera = async (
request: Extract<PublishRequest, { source: "camera" }>,
): Promise<void> => {
let deviceId = request.deviceId;
let facingMode: string | undefined;
let processor = request.processor;
let capture: Capture | null = null;
let processorStarted = false;
const constraints = (): MediaTrackConstraints => ({
deviceId: exact(deviceId),
facingMode,
width: request.capture?.resolution?.width,
height: request.capture?.resolution?.height,
frameRate: request.capture?.resolution?.frameRate,
});
const open = async (): Promise<Capture> => {
const raw = await getTrack({ video: constraints() });
if (processor === undefined) return { raw, out: raw };
const out = processorStarted
? await processor.restart(raw)
: await processor.init(raw);
processorStarted = true;
return { raw, out };
};
const close = (): void => {
capture?.raw.stop();
if (capture?.out !== capture?.raw) capture?.out.stop();
capture = null;
};
// Reopens the camera with the current settings, if it is on
const reopen = async (
track$: BehaviorSubject<MediaStreamTrack | null>,
): Promise<void> => {
if (capture === null) return;
close();
capture = await open();
track$.next(capture.out);
};
capture = await open();
addPublication("camera", "video", userMediaStream, capture.out, {
// Muting a camera releases it, so that its light goes off
setEnabled: async (enabled, track$) => {
if (enabled === (capture !== null)) return enabled;
if (!enabled) {
close();
track$.next(null);
return false;
}
try {
capture = await open();
track$.next(capture.out);
return true;
} catch (e) {
logger.warn("Could not reopen the camera", e);
return false;
}
},
setDevice: async (id, track$) => {
deviceId = id;
facingMode = undefined;
await reopen(track$);
},
switchFacingMode: async (track$) => {
const current = facingModeOf(track$.value);
if (current === undefined) return undefined;
deviceId = undefined;
facingMode = current === "user" ? "environment" : "user";
await reopen(track$);
return capture?.raw.getSettings().deviceId;
},
setProcessor: async (p, track$) => {
await processor?.destroy();
processor = p;
processorStarted = false;
if (capture === null) return;
const out = p ? await p.init(capture.raw) : capture.raw;
processorStarted = p !== undefined;
if (capture.out !== capture.raw) capture.out.stop();
capture = { raw: capture.raw, out };
track$.next(out);
},
stop: () => {
close();
void processor?.destroy();
},
});
};
const captureScreen = async (
request: Extract<PublishRequest, { source: "screenShare" }>,
): Promise<void> => {
const stream = await navigator.mediaDevices.getDisplayMedia({
video: {
width: request.capture?.resolution?.width,
height: request.capture?.resolution?.height,
frameRate: request.capture?.resolution?.frameRate,
},
audio: request.audio ?? true,
});
const [video] = stream.getVideoTracks();
const [audio] = stream.getAudioTracks();
const toggle = async (
enabled: boolean,
track$: BehaviorSubject<MediaStreamTrack | null>,
): Promise<boolean> => {
if (track$.value) track$.value.enabled = enabled;
return Promise.resolve(enabled);
};
addPublication("screenShare", "video", screenShareStream, video, {
setEnabled: toggle,
// The audio goes with its video
stop: () => {
audio?.stop();
teardowns.get("screenShareAudio")?.();
},
});
if (audio)
addPublication("screenShareAudio", "audio", screenShareStream, audio, {
setEnabled: toggle,
stop: () => {},
});
// Stopped from the browser's own controls
video.addEventListener("ended", () => void unpublish("screenShare"));
};
const capture = async (request: PublishRequest): Promise<void> => {
teardowns.get(request.source)?.();
switch (request.source) {
case "microphone":
return captureMicrophone(request);
case "camera":
return captureCamera(request);
case "screenShare":
return captureScreen(request);
}
};
const start = async (): Promise<void> => {
for (const request of desired.values())
await capture(request).catch((e: unknown) => {
logger.warn(`Could not capture ${request.source}`, e);
if (publishError$.value === null)
publishError$.next(toMatrixRTCError(e));
});
};
const setPublishing = (publish: boolean): void => {
publishing$.next(publish);
if (publish && !started$.value) {
started$.next(true);
void start();
}
};
const publish = async (
request: PublishRequest,
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack> => {
desired.set(request.source, request);
try {
// Before the start, the start captures everything desired itself
if (started$.value) await capture(request);
return await firstValueFrom(
publications$.pipe(
map((ps) => ps.find((p) => p.source === request.source)),
filter((p) => p !== undefined),
map((p) => p.mediaTrack),
),
);
} catch (e) {
desired.delete(request.source);
throw toMatrixRTCError(e);
}
};
const unpublish = async (source: MediaSource): Promise<void> => {
const published = source === "screenShareAudio" ? "screenShare" : source;
if (published === "unknown") return;
desired.delete(published);
teardowns.get(published)?.();
return Promise.resolve();
};
scope.onEnd(() => {
// Each teardown removes itself from the map
for (const teardown of Array.from(teardowns.values())) teardown();
});
const tracks$ = scope.behavior(
combineLatest([started$, publications$]).pipe(
map(([started, ps]) => (started ? ps.map((p) => p.mediaTrack) : null)),
),
);
const streamMetadata$ = scope.behavior(
publications$.pipe(
switchMap((ps) =>
ps.length === 0
? of([])
: combineLatest(
ps.map((p) =>
p.mediaTrack.muted$.pipe(map((muted) => ({ p, muted }))),
),
),
),
map((entries) => {
const metadata: StreamMetadata = {};
for (const { p, muted } of entries) {
const entry = (metadata[p.stream.id] ??= {
purpose:
p.stream === userMediaStream ? "m.usermedia" : "m.screenshare",
});
if (p.kind === "audio") entry.audio_muted = muted;
else entry.video_muted = muted;
}
return metadata;
}),
),
);
return {
connectionState$,
tracks$,
encryptionError$: NEVER,
setPublishing,
publishError$,
publish,
unpublish,
publications$,
publishing$,
streamMetadata$,
};
}
function exact(deviceId: string | undefined): ConstrainDOMString | undefined {
return deviceId === undefined ? undefined : { exact: deviceId };
}
async function getTrack(
constraints: MediaStreamConstraints,
): Promise<MediaStreamTrack> {
const stream = await navigator.mediaDevices.getUserMedia(constraints);
const [track] = stream.getTracks();
if (track === undefined) throw new MatrixRTCError("No track was captured");
return track;
}
function facingModeOf(
track: MediaStreamTrack | null,
): "user" | "environment" | undefined {
const mode = track?.getSettings().facingMode;
return mode === "user" || mode === "environment" ? mode : undefined;
}
@@ -0,0 +1,350 @@
# MSCXXXX: Full mesh transport for MatrixRTC
[MSC4143] introduces MatrixRTC. Concrete transports are defined in seperate proposals. [MSC4195] defines an
SFU based transport (livekit). This proposal defines a peer-to-peer transport.
In one sentence:\
Every subscriber opens a direct WebRTC connection to every publisher it wants to receive media from. No server-side infrastructure is
needed apart from the TURN servers homeservers already provide for [VoIP].
The transport only covers establishing WebRTC connections between two MatrixRTC members. Calls, ringing,
invites, call IDs and membership are all handled by [MSC4143] and the respective application.
[MSC4143]: https://github.com/matrix-org/matrix-spec-proposals/pull/4143
[MSC4195]: https://github.com/matrix-org/matrix-spec-proposals/pull/4195
[VoIP]: https://spec.matrix.org/v1.18/client-server-api/#voice-over-ip
## Proposal
### Transport type
A new transport type `m.full_mesh` is introduced. A member publishing on it adds the following object
to `transports.published` in its `m.rtc.member` event:
- `type` (required, string): `m.full_mesh`.
- `device_id` (required, string): The device ID of the publishing client. Subscribers send to-device
messages to this device. Needed for unencrypted rooms (We can only resolve the device id of room events
from the olm session info)
Members able to subscribe add `m.full_mesh` to `transports.can_subscribe`.
```json5
{
type: "m.rtc.member",
content: {
slot_id: "m.call#room",
member: { id: "xyzABCDEF10123", membership: "join" },
application: { type: "m.call" },
transports: {
published: [{ type: "m.full_mesh", device_id: "ABCDEFGHIJ" }],
can_subscribe: ["m.full_mesh", "m.livekit"],
},
sticky_key: "xyzABCDEF10123",
},
}
```
In encrypted rooms, clients MUST ignore an `m.full_mesh` transport whose `device_id` does not match the
device that sent the `m.rtc.member` event.
A member MUST NOT list more than one `m.full_mesh` object in `transports.published`. Servers don't need
to advertise `m.full_mesh` in `/_matrix/client/v1/rtc/transports`.
### Connection model
Each WebRTC connection is **unidirectional** and belongs to exactly one _(subscriber, publisher)_ pair of
MatrixRTC members:
- The **subscriber** starts the connection by asking the publisher for its media.
- The **publisher** is always the SDP offerer and only sends media. The subscriber only receives media.
- If two members subscribe to each other, they use two independent connections.
This lets transports be mixed per direction. Alice can subscribe to Bob over `m.full_mesh` while Bob
subscribes to Alice over `m.livekit`. Because only the publisher creates offers, there is no glare.
At most one connection exists per (subscriber `member.id`, publisher `member.id`) pair.
### TURN servers
Each client gets ICE servers from its own homeserver via the existing
[`GET /_matrix/client/v3/voip/turnServer`][turnServer] endpoint and passes `uris`, `username` and
`password` to every `RTCPeerConnection` it creates. Clients SHOULD fetch the credentials again before
`ttl` expires and use the new ones for new connections. If the response is empty, the homeserver has no
TURN server. Clients then use host and server-reflexive candidates only. They MUST NOT fall back to a
third-party STUN/TURN server without user consent.
[turnServer]: https://spec.matrix.org/v1.18/client-server-api/#get_matrixclientv3voipturnserver
### To-device messages
All signalling uses [to-device messages] sent to the single device of the target member. Messages MUST be
Olm-encrypted. Recipients MUST discard messages sent in cleartext.
[to-device messages]: https://spec.matrix.org/v1.18/client-server-api/#send-to-device-messaging
Every message contains these common properties:
- `room_id` (required, string): The room containing the slot.
- `slot_id` (required, string): The ID of the slot.
- `member_id` (required, string): The sender's `member.id`.
- `device_id` (required, string): The sender's device ID. The publisher replies to this device, even when
the subscriber doesn't publish `m.full_mesh` and so has no `device_id` in its transport. Clients often
can't get the Olm sending device of a to-device message from their SDK or the widget API, so the sender
states it here. This is the same approach as `claimed_device_id` in the reference implementation of
[MSC4143].
- `target_member_id` (required, string): The recipient's `member.id`.
- `connection_id` (required, string): A random identifier the subscriber generates for each subscription
attempt. All messages of the connection use it.
- `seq` (required, integer): Starts at `0` for each (`member_id`, `target_member_id`) pair and goes up by
`1` with every message sent from `member_id` to `target_member_id`. It is _not_ reset when a new
`connection_id` is used, so messages are ordered across reconnections too. To-device messages are not
guaranteed to arrive in order. Recipients MUST therefore process messages in `seq` order and buffer
them until any gaps are filled. If a gap is not filled within a short timeout (RECOMMENDED: 10
seconds), recipients MUST skip the missing messages and continue with the buffered ones.
Recipients MUST discard a message if any of these is true:
- `room_id` does not match the room of an active shared (between the sender and receiver) rtc session.
- `slot_id` does not match the slot of an active shared rtc session.
- `target_member_id` is not their own current `member.id`.
- No joined `m.rtc.member` event matches `member_id`.
- That event's sender is not the sender of the to-device message.
- `device_id` is not the device that event belongs to.
- The client knows the Olm sending device of the to-device message, and it differs from `device_id`.
- The message is not `m.rtc.full_mesh.subscribe` and its `connection_id` is not the current connection
for this member pair. This drops late messages belonging to a replaced or closed connection.
A subscribe request MAY arrive before the subscriber's `m.rtc.member` event. Publishers SHOULD keep such
a request for a short grace period (RECOMMENDED: 10 seconds) before discarding it.
#### `m.rtc.full_mesh.subscribe` (subscriber → publisher)
Asks the publisher to open a connection and send its media. It contains only the common properties.
#### `m.rtc.full_mesh.description` (both directions)
Carries an SDP offer (publisher → subscriber) or answer (subscriber → publisher).
- `description` (required, object): An [`RTCSessionDescriptionInit`].
- `type` (required, string): `offer` or `answer`.
- `sdp` (required, string): The SDP.
- `stream_metadata` (object): REQUIRED on offers. See [stream metadata](#stream-metadata).
```json5
{
room_id: "!room:example.org",
slot_id: "m.call#room",
member_id: "bobMemberId",
device_id: "BOBDEVICE",
target_member_id: "aliceMemberId",
connection_id: "c0nn3ct10n1d",
seq: 0,
description: { type: "offer", sdp: "v=0\r\n..." },
stream_metadata: {
streamIdA: {
purpose: "m.usermedia",
audio_muted: false,
video_muted: true,
},
},
}
```
#### `m.rtc.full_mesh.candidates` (both directions)
Carries trickled ICE candidates.
- `candidates` (required, array): An array of [`RTCIceCandidateInit`] objects (`candidate`, `sdpMid`,
`sdpMLineIndex`, `usernameFragment`). A `candidate` set to the empty string signals end-of-candidates.
Including `usernameFragment` lets the WebRTC stack discard candidates belonging to an outdated
description.
Clients SHOULD collect candidates for a short time (for example 200ms) and send them in one message
rather than one message per candidate.
Candidates can't overtake the description they belong to. A client only gathers candidates after it
has set its local description, so its candidates always have a higher `seq` than that description.
Recipients MUST still wait for `setRemoteDescription` to complete before adding candidates.
#### `m.rtc.full_mesh.stream_metadata` (publisher → subscriber)
Updates stream metadata without renegotiating, for example when a stream is muted.
- `stream_metadata` (required, object): The full metadata for this connection. It replaces the previous
metadata.
#### `m.rtc.full_mesh.close` (both directions)
Closes the connection. The recipient closes the corresponding `RTCPeerConnection`.
- `reason` (string): An optional human-readable reason.
#### Stream metadata
`stream_metadata` maps each WebRTC `MediaStream` ID in the SDP (the `msid` stream ID) to an object:
- `purpose` (required, string): `m.usermedia` or `m.screenshare`. Subscribers SHOULD ignore streams with
an unknown purpose.
- `audio_muted` (boolean): Whether the stream's audio is muted. Defaults to `false`.
- `video_muted` (boolean): Whether the stream's video is muted. Defaults to `false`.
Following MSCs MAY define further properties.
[`RTCSessionDescriptionInit`]: https://www.w3.org/TR/webrtc/#dom-rtcsessiondescriptioninit
[`RTCIceCandidateInit`]: https://www.w3.org/TR/webrtc/#dom-rtcicecandidateinit
### Procedures
#### Subscribing
A subscriber (Alice) that wants media from a joined member (Bob) whose `transports.published` contains an
`m.full_mesh` object:
1. Alice creates a new `connection_id` (random) and sends `m.rtc.full_mesh.subscribe` to Bob's `device_id`.
2. Bob validates the request as described above. If Bob doesn't publish `m.full_mesh`, or the request is
invalid, Bob MAY reply with `m.rtc.full_mesh.close`. If Bob already has a connection for Alice's
`member.id`, he closes it and replaces it with the new one.
3. Bob creates an `RTCPeerConnection` using his own ICE servers and adds his current tracks as `sendonly`.
He then creates an offer, sets it as the local description and sends it with `stream_metadata` in an
`m.rtc.full_mesh.description` message.
4. Alice creates an `RTCPeerConnection` using her own ICE servers and sets the offer as the remote
description. She then creates an answer (`recvonly`), sets it as the local description and sends it
in an `m.rtc.full_mesh.description` message.
5. Both sides send `m.rtc.full_mesh.candidates` as they gather candidates.
```mermaid
sequenceDiagram
participant A as Alice (subscriber)
participant B as Bob (publisher)
A-->>B: subscribe
B-->>A: description (offer + stream_metadata)
A-->>B: description (answer)
A-->>B: candidates
B-->>A: candidates
B->>A: media (DTLS-SRTP)
```
#### Renegotiation
When the publisher's tracks change (for example when a screenshare is added or removed), the publisher
sends a new offer with the same `connection_id`. The subscriber answers it. The publisher MUST NOT send
a new offer while a previous one is still unanswered (signalling state not `stable`). Instead, it waits
for the answer and combines all track changes made in the meantime into one new offer. When only mute state
changes, the publisher sends `m.rtc.full_mesh.stream_metadata` instead.
#### Closing
- The subscriber sends `m.rtc.full_mesh.close` when it no longer wants the media, or when the publisher
leaves the slot or stops publishing `m.full_mesh`.
- The publisher sends `m.rtc.full_mesh.close` when it stops publishing `m.full_mesh`, or when the
subscriber is no longer considered joined as per [MSC4143]. Publishers MUST stop sending media to
members that are no longer joined.
- Leaving the slot closes all of a member's connections.
#### Failure
If a connection's ICE state becomes `failed`, the subscriber MAY send `m.rtc.full_mesh.close` and then
subscribe again with a new `connection_id`. It SHOULD use exponential backoff between attempts.
Publishers don't retry.
### End-to-end encryption
WebRTC always encrypts media with DTLS-SRTP. Each side's DTLS certificate fingerprint is part of its SDP.
The SDP is carried in Olm-encrypted to-device messages, and recipients check that the sending device
matches the member's `m.rtc.member` event. As a result, the fingerprints are authenticated end to end
between the two member devices, and the media is end-to-end encrypted. TURN servers relay ciphertext only.
DTLS uses an ephemeral key exchange, so every connection also has forward secrecy.
`m.full_mesh` therefore doesn't use the `m.per_member` key material from [MSC4143], and its required key
length is 0 bytes. A client whose published transports are all `m.full_mesh` doesn't need to send
`m.rtc.encryption_key` messages. Media received over a connection that was set up as described above
counts as encrypted for the purposes of [MSC4143].
## Potential issues
### Scalability
Every publisher uploads one copy of its media per subscriber, and the number of connections grows
quadratically with the number of members. The transport is only suitable for small sessions. Clients MAY
refuse subscriptions beyond a limit they choose.
### To-device latency
Connection setup needs about two to-device round trips, possibly over federation. This is slower than
connecting to an SFU.
### Twice the connections for bidirectional media
Using one connection per direction doubles the number of ICE and DTLS handshakes between two members who
both publish and subscribe. In return, transports can be mixed and there is no glare.
## Alternatives
### One bidirectional connection per pair
[MSC3401] used one connection per pair of members, with a tie-break rule for who calls whom. This halves
the number of connections. However, it ties both directions to the same transport and needs glare
handling for renegotiation.
[MSC3401]: https://github.com/matrix-org/matrix-spec-proposals/pull/3401
### Subscriber as offerer
The subscriber could send an offer with `recvonly` transceivers directly, saving one message. However, the
subscriber doesn't know how many streams the publisher has, so the publisher would need to make offers
for renegotiation anyway. Making the publisher the only offerer keeps a single code path.
### Reusing `m.call.*` events
The legacy `m.call.invite`, `m.call.answer` and similar events bring call IDs, party IDs, invite lifetimes
and ringing semantics. MatrixRTC already handles all of these, so the events were not reused.
### Using `m.per_member` keys with SFrame
Media could also be encrypted with the [MSC4143] key material on top of DTLS-SRTP. This adds no security
over authenticated DTLS-SRTP, so it is not required.
## Security considerations
### IP address exposure
Peers learn each other's IP addresses from ICE candidates. Clients MAY set `iceTransportPolicy: "relay"`
so that only their homeserver's TURN server is exposed, at the cost of more latency. Candidates are only
sent through Olm-encrypted to-device messages, and only to members that are joined to the slot.
### Unencrypted rooms
In unencrypted rooms, `m.rtc.member` events are not authenticated. A malicious homeserver could forge a
member event with a `device_id` it controls and receive media. Signalling itself stays Olm-encrypted, so
existing connections can't be tampered with. Clients SHOULD only send to devices that are cross-signed
by their owner, as described in [MSC4153].
[MSC4153]: https://github.com/matrix-org/matrix-spec-proposals/pull/4153
### Claimed device IDs
A recipient that can't check `device_id` against the Olm sending device only knows that the message
comes from the right user. Another device of that same user could pretend to be the device of the
member and receive its media. Only that user's own devices can do this, the same as for `claimed_device_id`
in [MSC4143]'s reference implementation. Clients SHOULD check the Olm sending device wherever their SDK
exposes it.
### Removed members
Publishers stop sending media when a subscriber is no longer joined. This way, redacting member events
or removing someone from the room also cuts their media.
## Unstable prefix
| Stable identifier | Purpose | Unstable identifier |
| --------------------------------- | -------------------- | -------------------------------------------------- |
| `m.full_mesh` | Transport type | `org.matrix.mscXXXX.full_mesh` |
| `m.rtc.full_mesh.subscribe` | To-device event type | `org.matrix.mscXXXX.rtc.full_mesh.subscribe` |
| `m.rtc.full_mesh.description` | To-device event type | `org.matrix.mscXXXX.rtc.full_mesh.description` |
| `m.rtc.full_mesh.candidates` | To-device event type | `org.matrix.mscXXXX.rtc.full_mesh.candidates` |
| `m.rtc.full_mesh.stream_metadata` | To-device event type | `org.matrix.mscXXXX.rtc.full_mesh.stream_metadata` |
| `m.rtc.full_mesh.close` | To-device event type | `org.matrix.mscXXXX.rtc.full_mesh.close` |
## Dependencies
This proposal depends on [MSC4143].
@@ -0,0 +1,215 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import {
combineLatest,
distinctUntilChanged,
map,
Observable,
of,
scan,
switchMap,
timer,
} from "rxjs";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import {
type AudioMediaTrack,
type MediaSource,
type MediaStreamStats,
type MediaTrack,
type VideoMediaTrack,
} from "../../media-api";
export interface TrackProps {
source: MediaSource;
kind: "audio" | "video";
id: string;
/** The track behind it; may change, as on a device switch, or be null while off. */
track$: Behavior<MediaStreamTrack | null>;
muted$: Behavior<boolean>;
/** Statistics of the RTP stream carrying it, if there is one. */
stats: () => Promise<MediaStreamStats>;
/** Where attached elements play; only for tracks that are played. */
audioOutputDeviceId$?: Behavior<string | undefined>;
}
/** Above this level a track turns active; below `INACTIVE_LEVEL` for `INACTIVE_AFTER_MS` it turns inactive. */
const ACTIVE_LEVEL = 0.04;
const INACTIVE_LEVEL = 0.02;
const LEVEL_INTERVAL_MS = 100;
const INACTIVE_AFTER_MS = 500;
let levelContext: AudioContext | undefined;
/**
* A `MediaStreamTrack` as a `MediaTrack`: the same object for the life of a
* publication, with every attached element following the track behind it.
*/
export function createFullMeshMediaTrack(
scope: ObservableScope,
props: TrackProps,
): AudioMediaTrack | VideoMediaTrack {
const { source, kind, id, track$, muted$, stats, audioOutputDeviceId$ } =
props;
const attached = new Set<HTMLMediaElement>();
let audioContext: AudioContext | undefined;
let audioPlugins: AudioNode[] = [];
let volume = 1;
let webAudio: { nodes: AudioNode[]; gain: GainNode } | undefined;
const routeWebAudio = (track: MediaStreamTrack | null): void => {
for (const node of webAudio?.nodes ?? []) node.disconnect();
webAudio = undefined;
if (audioContext === undefined || track === null) return;
const input = audioContext.createMediaStreamSource(
new MediaStream([track]),
);
const gain = audioContext.createGain();
gain.gain.value = volume;
const chain: AudioNode[] = [input, ...audioPlugins, gain];
for (let i = 0; i < chain.length - 1; i++) chain[i].connect(chain[i + 1]);
gain.connect(audioContext.destination);
webAudio = { nodes: chain, gain };
};
const apply = (element: HTMLMediaElement): void => {
const track = track$.value;
element.srcObject = track ? new MediaStream([track]) : null;
if (kind === "audio") {
// Played through Web Audio instead, where it is routed there
element.muted = audioContext !== undefined;
element.volume = volume;
const sinkId = audioOutputDeviceId$?.value;
if (sinkId !== undefined && "setSinkId" in element)
void element.setSinkId(sinkId).catch(() => {});
}
if (element.paused) void element.play().catch(() => {});
};
track$.pipe(scope.bind()).subscribe((track) => {
for (const element of attached) apply(element);
if (kind === "audio") routeWebAudio(track);
});
audioOutputDeviceId$?.pipe(scope.bind()).subscribe(() => {
for (const element of attached) apply(element);
});
scope.onEnd(() => {
for (const element of attached) element.srcObject = null;
attached.clear();
routeWebAudio(null);
});
const base: MediaTrack = {
source,
kind,
id,
muted$,
// DTLS-SRTP, its fingerprints exchanged over Olm
encrypted: true,
stats$: (intervalMs): Observable<MediaStreamStats> =>
timer(0, intervalMs).pipe(
switchMap(async () => stats()),
scope.bind(),
),
attach: (element) => {
if (attached.has(element)) return;
attached.add(element);
apply(element);
},
detach: (element) => {
if (!attached.delete(element)) return;
element.srcObject = null;
},
};
if (kind === "video") return { ...base, kind: "video" };
const audioLevel$ = scope.behavior(
combineLatest([track$, muted$]).pipe(
switchMap(([track, muted]) =>
track === null || muted ? of(0) : level$(track),
),
),
0,
);
const isActive$ = scope.behavior(
audioLevel$.pipe(
scan(
({ active, quietMs }, level) => {
if (level >= ACTIVE_LEVEL) return { active: true, quietMs: 0 };
if (level >= INACTIVE_LEVEL) return { active, quietMs: 0 };
const quiet = quietMs + LEVEL_INTERVAL_MS;
return {
active: active && quiet < INACTIVE_AFTER_MS,
quietMs: quiet,
};
},
{ active: false, quietMs: 0 },
),
map(({ active }) => active),
distinctUntilChanged(),
),
false,
);
return {
...base,
kind: "audio",
audioLevel$,
isActive$,
setAudioContext: (ctx, plugins = []) => {
audioContext = ctx;
audioPlugins = plugins;
routeWebAudio(track$.value);
for (const element of attached) apply(element);
},
setVolume: (v) => {
volume = v;
if (webAudio) webAudio.gain.gain.value = v;
for (const element of attached) element.volume = v;
},
};
}
/** The RMS level of an audio track, 0 to 1, sampled while subscribed. */
function level$(track: MediaStreamTrack): Observable<number> {
return new Observable<number>((subscriber) => {
levelContext ??= new AudioContext();
const ctx = levelContext;
const input = ctx.createMediaStreamSource(new MediaStream([track]));
const analyser = ctx.createAnalyser();
analyser.fftSize = 512;
input.connect(analyser);
const samples = new Float32Array(analyser.fftSize);
const sub = timer(0, LEVEL_INTERVAL_MS).subscribe(() => {
analyser.getFloatTimeDomainData(samples);
let sum = 0;
for (const s of samples) sum += s * s;
subscriber.next(Math.min(1, Math.sqrt(sum / samples.length) * 4));
});
return () => {
sub.unsubscribe();
input.disconnect();
};
});
}
/** The inbound or outbound RTP statistics of one track on a peer connection. */
export async function rtpStats(
pc: RTCPeerConnection | undefined,
track: MediaStreamTrack | null,
direction: "inbound-rtp" | "outbound-rtp",
): Promise<MediaStreamStats> {
if (pc === undefined || track === null) return undefined;
const report = await pc.getStats(track);
for (const stats of report.values())
if ((stats as RTCStats).type === direction)
return stats as RTCInboundRtpStreamStats & RTCOutboundRtpStreamStats;
return undefined;
}
@@ -0,0 +1,153 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import { ObservableScope } from "../../reactive/ObservableScope";
import { MediaConnectionState } from "../api";
import {
CANDIDATE_BATCH_MS,
createPeerConnectionBase,
type PeerConnectionBase,
} from "./PeerConnection";
import { SignallingEventType } from "./signalling";
import {
type FakeSignalling,
fakeSignalling,
fromPeer,
installWebRtcFakes,
lastPeerConnection,
peer,
settle,
} from "./test";
describe("createPeerConnectionBase", () => {
let scope: ObservableScope;
let signalling: FakeSignalling;
let base: PeerConnectionBase;
beforeEach(() => {
vi.useFakeTimers();
installWebRtcFakes();
scope = new ObservableScope();
signalling = fakeSignalling();
base = createPeerConnectionBase({
scope,
peer,
connectionId: "c1",
signalling,
iceServers: [{ urls: ["turn:example.org"] }],
logger,
});
});
afterEach(() => {
scope.end();
vi.useRealTimers();
vi.unstubAllGlobals();
});
it("uses the ICE servers it was given", () => {
expect(lastPeerConnection().config.iceServers).toEqual([
{ urls: ["turn:example.org"] },
]);
});
it("sends the candidates gathered within a batch in one message", () => {
const pc = lastPeerConnection();
pc.gather({ candidate: "a", sdpMid: "0", sdpMLineIndex: 0 });
pc.gather({ candidate: "b", sdpMid: "0", sdpMLineIndex: 0 });
pc.gather(null);
expect(signalling.send).not.toHaveBeenCalled();
vi.advanceTimersByTime(CANDIDATE_BATCH_MS);
expect(signalling.sent()).toEqual([
{
type: SignallingEventType.Candidates,
candidates: [
{ candidate: "a", sdpMid: "0", sdpMLineIndex: 0 },
{ candidate: "b", sdpMid: "0", sdpMLineIndex: 0 },
// The end of the gathering
{ candidate: "", sdpMid: null, sdpMLineIndex: null },
],
},
]);
expect(signalling.sentConnectionIds()).toEqual(["c1"]);
});
it("drops pending candidates when it closes", () => {
lastPeerConnection().gather({ candidate: "a" });
base.close("bye", false);
vi.advanceTimersByTime(CANDIDATE_BATCH_MS);
expect(signalling.send).not.toHaveBeenCalled();
});
it("runs queued operations in order, past a failing one", async () => {
const order: string[] = [];
base.enqueue(async () => {
order.push("first");
return Promise.reject(new Error("failed"));
});
base.enqueue(async () => {
order.push("second");
return Promise.resolve();
});
await settle();
expect(order).toEqual(["first", "second"]);
});
it("applies received candidates, ignoring one that fails", async () => {
base.addCandidates(
fromPeer("c1", {
type: SignallingEventType.Candidates,
candidates: [{ candidate: "bad" }, { candidate: "good" }],
}) as Parameters<PeerConnectionBase["addCandidates"]>[0],
);
await settle();
expect(lastPeerConnection().addedCandidates).toEqual([
{ candidate: "good" },
]);
});
it("tells the peer once when it closes", () => {
base.close("done", true);
base.close("again", true);
expect(lastPeerConnection().closed).toBe(true);
expect(signalling.sent()).toEqual([
{ type: SignallingEventType.Close, reason: "done" },
]);
});
it("doesn't tell a peer that closed first", () => {
base.close("theirs", false);
expect(lastPeerConnection().closed).toBe(true);
expect(signalling.send).not.toHaveBeenCalled();
});
it("closes when its scope ends", () => {
scope.end();
expect(lastPeerConnection().closed).toBe(true);
expect(signalling.sent()).toEqual([
{ type: SignallingEventType.Close, reason: "ended" },
]);
});
it("maps the connection state", () => {
const pc = lastPeerConnection();
expect(base.state$.value).toBe(MediaConnectionState.Initialized);
pc.setConnectionState("connecting");
expect(base.state$.value).toBe(MediaConnectionState.Connecting);
pc.setConnectionState("connected");
expect(base.state$.value).toBe(MediaConnectionState.Connected);
pc.setConnectionState("disconnected");
expect(base.state$.value).toBe(MediaConnectionState.Reconnecting);
pc.setConnectionState("failed");
expect(base.state$.value).toBeInstanceOf(Error);
pc.setConnectionState("closed");
expect(base.state$.value).toBe(MediaConnectionState.Stopped);
});
});
@@ -0,0 +1,136 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
/** What the publisher and the subscriber side of a connection share. */
import { type Logger } from "matrix-js-sdk/lib/logger";
import { BehaviorSubject, fromEvent, map, startWith } from "rxjs";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { MediaConnectionState } from "../api";
import {
type Peer,
type Signalling,
SignallingEventType,
type SignallingMessage,
} from "./signalling";
/** How long trickled candidates are collected before they are sent in one message. */
export const CANDIDATE_BATCH_MS = 200;
export interface PeerConnectionBase {
readonly connectionId: string;
readonly pc: RTCPeerConnection;
readonly state$: Behavior<MediaConnectionState | Error>;
/** Runs `op` after every operation queued before it, so that SDP and candidates apply in order. */
enqueue(op: () => Promise<void>): void;
/** Applies a received batch of candidates. */
addCandidates(
message: SignallingMessage & {
type: typeof SignallingEventType.Candidates;
},
): void;
/** Closes the peer connection, telling the peer unless it closed first. */
close(reason: string | undefined, notifyPeer: boolean): void;
}
export function createPeerConnectionBase({
scope,
peer,
connectionId,
signalling,
iceServers,
logger,
}: {
scope: ObservableScope;
peer: Peer;
connectionId: string;
signalling: Signalling;
iceServers: RTCIceServer[];
logger: Logger;
}): PeerConnectionBase {
const pc = new RTCPeerConnection({ iceServers });
const closed$ = new BehaviorSubject(false);
let queue = Promise.resolve();
const enqueue = (op: () => Promise<void>): void => {
queue = queue.then(op).catch((e: unknown) => {
logger.warn("Signalling operation failed", e);
});
};
// Trickled, in batches
let pending: RTCIceCandidateInit[] = [];
let batchTimer: ReturnType<typeof setTimeout> | undefined;
const flushCandidates = (): void => {
batchTimer = undefined;
if (pending.length === 0 || closed$.value) return;
signalling.send(peer, connectionId, {
type: SignallingEventType.Candidates,
candidates: pending,
});
pending = [];
};
pc.addEventListener("icecandidate", ({ candidate }) => {
// A null candidate ends the gathering: an empty one says so on the wire
pending.push(
candidate
? candidate.toJSON()
: { candidate: "", sdpMid: null, sdpMLineIndex: null },
);
batchTimer ??= setTimeout(flushCandidates, CANDIDATE_BATCH_MS);
});
const addCandidates: PeerConnectionBase["addCandidates"] = ({ candidates }) =>
enqueue(async () => {
for (const candidate of candidates)
await pc
.addIceCandidate(candidate)
// A candidate of an outdated description, or a bad one: not fatal
.catch((e: unknown) => logger.debug("Ignored a candidate", e));
});
const close: PeerConnectionBase["close"] = (reason, notifyPeer) => {
if (closed$.value) return;
clearTimeout(batchTimer);
if (notifyPeer)
signalling.send(peer, connectionId, {
type: SignallingEventType.Close,
reason,
});
pc.close();
closed$.next(true);
};
scope.onEnd(() => close("ended", true));
const state$ = scope.behavior(
fromEvent(pc, "connectionstatechange").pipe(
startWith(null),
map(() => toState(pc.connectionState)),
),
);
return { connectionId, pc, state$, enqueue, addCandidates, close };
}
function toState(state: RTCPeerConnectionState): MediaConnectionState | Error {
switch (state) {
case "new":
return MediaConnectionState.Initialized;
case "connecting":
return MediaConnectionState.Connecting;
case "connected":
return MediaConnectionState.Connected;
case "disconnected":
return MediaConnectionState.Reconnecting;
case "failed":
return new Error("The peer connection failed");
case "closed":
return MediaConnectionState.Stopped;
}
}
@@ -0,0 +1,200 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type FullMeshLocalMedia } from "./LocalMedia";
import {
createPublisherConnection,
type PublisherConnection,
} from "./PublisherConnection";
import { SignallingEventType } from "./signalling";
import {
type FakeSignalling,
fakeLocalMedia,
fakePublication,
fakeSignalling,
fakeTrack,
fromPeer,
installWebRtcFakes,
lastPeerConnection,
peer,
settle,
} from "./test";
describe("createPublisherConnection", () => {
let scope: ObservableScope;
let signalling: FakeSignalling;
let local: ReturnType<typeof fakeLocalMedia>;
function create(): PublisherConnection {
return createPublisherConnection({
scope,
peer,
connectionId: "c1",
signalling,
iceServers: [],
local: local as unknown as FullMeshLocalMedia,
logger,
});
}
const answer = (): ReturnType<typeof fromPeer> =>
fromPeer("c1", {
type: SignallingEventType.Description,
description: { type: "answer", sdp: "answer" },
});
beforeEach(() => {
installWebRtcFakes();
scope = new ObservableScope();
signalling = fakeSignalling();
local = fakeLocalMedia();
});
afterEach(() => {
scope.end();
vi.unstubAllGlobals();
});
it("offers right away, even without tracks, with the stream metadata", async () => {
local.streamMetadata$.next({ s: { purpose: "m.usermedia" } });
create();
await settle();
expect(signalling.sent(SignallingEventType.Description)).toEqual([
{
type: SignallingEventType.Description,
description: { type: "offer", sdp: "offer-1" },
stream_metadata: { s: { purpose: "m.usermedia" } },
},
]);
});
it("sends each publication on a sendonly transceiver in its stream", async () => {
const mic = fakePublication("audio");
const camera = fakePublication("video", mic.stream);
local.publications$.next([mic, camera]);
create();
await settle();
const [a, v] = lastPeerConnection().transceivers;
expect(a).toMatchObject({ kind: "audio", direction: "sendonly" });
expect(a.streams).toEqual([mic.stream]);
expect(a.sender.track).toBe(mic.track$.value);
expect(v).toMatchObject({ kind: "video", direction: "sendonly" });
expect(v.sender.track).toBe(camera.track$.value);
// One offer for both
expect(signalling.sent(SignallingEventType.Description)).toHaveLength(1);
});
it("follows a track change without renegotiating", async () => {
const camera = fakePublication("video");
local.publications$.next([camera]);
create();
await settle();
const next = fakeTrack("video");
camera.track$.next(next);
expect(lastPeerConnection().transceivers[0].sender.track).toBe(next);
expect(signalling.sent(SignallingEventType.Description)).toHaveLength(1);
});
it("sends nothing while not publishing", async () => {
const camera = fakePublication("video");
local.publications$.next([camera]);
local.publishing$.next(false);
create();
await settle();
const { sender } = lastPeerConnection().transceivers[0];
expect(sender.track).toBeNull();
local.publishing$.next(true);
expect(sender.track).toBe(camera.track$.value);
});
it("holds changes while an offer is unanswered, then offers once for all", async () => {
const connection = create();
const pc = lastPeerConnection();
await settle();
expect(pc.offers).toBe(1);
local.publications$.next([fakePublication("audio")]);
local.publications$.next([
...local.publications$.value,
fakePublication("video"),
]);
await settle();
expect(pc.offers).toBe(1);
connection.handle(answer());
await settle();
expect(pc.offers).toBe(2);
expect(pc.transceivers).toHaveLength(2);
// Nothing changed since: the next answer doesn't lead to another offer
connection.handle(answer());
await settle();
expect(pc.offers).toBe(2);
});
it("stops the transceiver of a removed publication and renegotiates", async () => {
const camera = fakePublication("video");
local.publications$.next([camera]);
const connection = create();
await settle();
connection.handle(answer());
await settle();
local.publications$.next([]);
await settle();
const pc = lastPeerConnection();
expect(pc.transceivers[0].stopped).toBe(true);
expect(pc.offers).toBe(2);
});
it("signals mute changes without renegotiating", async () => {
create();
await settle();
local.streamMetadata$.next({
s: { purpose: "m.usermedia", audio_muted: true },
});
expect(signalling.sent(SignallingEventType.StreamMetadata)).toEqual([
{
type: SignallingEventType.StreamMetadata,
stream_metadata: { s: { purpose: "m.usermedia", audio_muted: true } },
},
]);
expect(lastPeerConnection().offers).toBe(1);
});
it("applies the subscriber's candidates", async () => {
const connection = create();
await settle();
connection.handle(answer());
connection.handle(
fromPeer("c1", {
type: SignallingEventType.Candidates,
candidates: [{ candidate: "x" }],
}),
);
await settle();
expect(lastPeerConnection().addedCandidates).toEqual([{ candidate: "x" }]);
});
it("ends without answering when the subscriber closes", async () => {
const connection = create();
await settle();
connection.handle(
fromPeer("c1", { type: SignallingEventType.Close, reason: "bye" }),
);
expect(lastPeerConnection().closed).toBe(true);
expect(signalling.sent(SignallingEventType.Close)).toEqual([]);
// Its scope ended: no more metadata goes out
local.streamMetadata$.next({ s: { purpose: "m.usermedia" } });
expect(signalling.sent(SignallingEventType.StreamMetadata)).toEqual([]);
});
});
@@ -0,0 +1,157 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type Logger } from "matrix-js-sdk/lib/logger";
import { combineLatest, skip, type Subscription } from "rxjs";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { type FullMeshLocalMedia, type LocalPublication } from "./LocalMedia";
import {
createPeerConnectionBase,
type PeerConnectionBase,
} from "./PeerConnection";
import {
type Peer,
type Signalling,
SignallingEventType,
type SignallingMessage,
} from "./signalling";
export interface PublisherConnection extends PeerConnectionBase {
/** A message of this connection from the subscriber. */
handle(message: SignallingMessage): void;
}
/**
* Our side of a connection a subscriber asked for: we only send, and we are
* the only side that makes offers, so there is no glare.
*/
export function createPublisherConnection({
scope,
peer,
connectionId,
signalling,
iceServers,
local,
logger: parentLogger,
}: {
scope: ObservableScope;
peer: Peer;
connectionId: string;
signalling: Signalling;
iceServers: RTCIceServer[];
local: FullMeshLocalMedia;
logger: Logger;
}): PublisherConnection {
const logger = parentLogger.getChild(
`[Publisher → ${peer.memberId} ${connectionId}]`,
);
const base = createPeerConnectionBase({
scope,
peer,
connectionId,
signalling,
iceServers,
logger,
});
const { pc, enqueue } = base;
// A new offer has to wait while one is unanswered; changes made meanwhile
// go out together in the next one
let offerPending = false;
let renegotiate = false;
const negotiate = (): void => {
if (offerPending) {
renegotiate = true;
return;
}
offerPending = true;
renegotiate = false;
enqueue(async () => {
const offer = await pc.createOffer();
await pc.setLocalDescription(offer);
signalling.send(peer, connectionId, {
type: SignallingEventType.Description,
description: { type: "offer", sdp: pc.localDescription!.sdp },
stream_metadata: local.streamMetadata$.value,
});
});
};
// One sendonly transceiver per publication, for as long as it exists
const transceivers = new Map<
LocalPublication,
{ transceiver: RTCRtpTransceiver; tracks: Subscription }
>();
scope.onEnd(() => {
for (const { tracks } of transceivers.values()) tracks.unsubscribe();
});
local.publications$.pipe(scope.bind()).subscribe((publications) => {
let changed = false;
for (const [publication, { transceiver, tracks }] of transceivers)
if (!publications.includes(publication)) {
tracks.unsubscribe();
if (pc.signalingState !== "closed") transceiver.stop();
transceivers.delete(publication);
changed = true;
}
for (const publication of publications)
if (!transceivers.has(publication)) {
const transceiver = pc.addTransceiver(publication.kind, {
direction: "sendonly",
streams: [publication.stream],
});
// Paused while not publishing, so that the local preview stays live
const tracks = combineLatest([
publication.track$,
local.publishing$,
]).subscribe(([track, publishing]) => {
void transceiver.sender
.replaceTrack(publishing ? track : null)
.catch((e: unknown) => logger.warn("replaceTrack failed", e));
});
transceivers.set(publication, { transceiver, tracks });
changed = true;
}
if (changed) negotiate();
});
// Even without tracks: the subscriber learns that we are there
if (transceivers.size === 0) negotiate();
// Mute changes without renegotiating; the offer carries the state at its time
local.streamMetadata$.pipe(skip(1), scope.bind()).subscribe((metadata) =>
signalling.send(peer, connectionId, {
type: SignallingEventType.StreamMetadata,
stream_metadata: metadata,
}),
);
const handle = (message: SignallingMessage): void => {
switch (message.type) {
case SignallingEventType.Description:
if (message.description.type !== "answer") return;
enqueue(async () => {
await pc.setRemoteDescription(message.description);
offerPending = false;
if (renegotiate) negotiate();
});
break;
case SignallingEventType.Candidates:
base.addCandidates(message);
break;
case SignallingEventType.Close:
logger.info(`Closed by the subscriber: ${message.reason}`);
base.close(message.reason, false);
scope.end();
break;
default:
logger.debug(`Ignoring ${message.type}`);
}
};
return { ...base, handle };
}
@@ -0,0 +1,214 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import { constant } from "../../reactive/Behavior";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type StreamMetadata, SignallingEventType } from "./signalling";
import {
createSubscriberConnection,
type SubscriberConnection,
} from "./SubscriberConnection";
import {
type FakeSignalling,
fakeSignalling,
fromPeer,
installWebRtcFakes,
lastPeerConnection,
peer,
settle,
} from "./test";
describe("createSubscriberConnection", () => {
let scope: ObservableScope;
let signalling: FakeSignalling;
let subscriber: SubscriberConnection;
const offer = (
connectionId: string,
metadata: StreamMetadata,
): ReturnType<typeof fromPeer> =>
fromPeer(connectionId, {
type: SignallingEventType.Description,
description: { type: "offer", sdp: "offer" },
stream_metadata: metadata,
});
/** The connection id of the current attempt. */
const current = (): string => {
const base = subscriber.current$.value;
if (base === null) throw new Error("No current attempt");
return base.connectionId;
};
beforeEach(async () => {
vi.useFakeTimers();
installWebRtcFakes();
scope = new ObservableScope();
signalling = fakeSignalling();
subscriber = createSubscriberConnection({
scope,
peer,
signalling,
iceServers: async () => Promise.resolve([{ urls: ["turn:x"] }]),
audioOutputDeviceId$: constant(undefined),
logger,
});
await settle();
});
afterEach(() => {
scope.end();
vi.useRealTimers();
vi.unstubAllGlobals();
});
it("asks the publisher for its media on a fresh connection", () => {
expect(signalling.sent()).toEqual([
{ type: SignallingEventType.Subscribe },
]);
expect(signalling.sentConnectionIds()).toEqual([current()]);
expect(lastPeerConnection().config.iceServers).toEqual([
{ urls: ["turn:x"] },
]);
});
it("answers an offer and shows the tracks it brings", async () => {
expect(subscriber.tracks$.value).toBeNull();
const pc = lastPeerConnection();
pc.receive("video", "user");
pc.receive("video", "screen");
subscriber.handle(
offer(current(), {
user: { purpose: "m.usermedia", video_muted: true },
screen: { purpose: "m.screenshare" },
}),
);
await settle();
expect(pc.remoteDescription).toEqual({ type: "offer", sdp: "offer" });
expect(signalling.sent(SignallingEventType.Description)).toEqual([
{
type: SignallingEventType.Description,
description: { type: "answer", sdp: "answer" },
},
]);
const tracks = subscriber.tracks$.value!;
expect(tracks.map((t) => [t.source, t.kind, t.muted$.value])).toEqual([
["camera", "video", true],
["screenShare", "video", false],
]);
});
it("names audio by its stream's purpose", async () => {
const pc = lastPeerConnection();
pc.receive("audio", "user");
pc.receive("audio", "screen");
subscriber.handle(
offer(current(), {
user: { purpose: "m.usermedia" },
screen: { purpose: "m.screenshare" },
}),
);
await settle();
expect(subscriber.tracks$.value!.map((t) => t.source)).toEqual([
"microphone",
"screenShareAudio",
]);
});
it("leaves out streams without metadata or with an unknown purpose", async () => {
const pc = lastPeerConnection();
pc.receive("video", "unknown");
pc.receive("video", "other");
subscriber.handle(
offer(current(), {
other: { purpose: "m.whiteboard" as "m.usermedia" },
}),
);
await settle();
expect(subscriber.tracks$.value).toEqual([]);
});
it("follows mute changes, keeping the same track", async () => {
lastPeerConnection().receive("video", "user");
subscriber.handle(offer(current(), { user: { purpose: "m.usermedia" } }));
await settle();
const [track] = subscriber.tracks$.value!;
expect(track.muted$.value).toBe(false);
subscriber.handle(
fromPeer(current(), {
type: SignallingEventType.StreamMetadata,
stream_metadata: {
user: { purpose: "m.usermedia", video_muted: true },
},
}),
);
await settle();
expect(subscriber.tracks$.value).toEqual([track]);
expect(track.muted$.value).toBe(true);
});
it("drops a track whose transceiver no longer receives", async () => {
const pc = lastPeerConnection();
const transceiver = pc.receive("video", "user");
subscriber.handle(offer(current(), { user: { purpose: "m.usermedia" } }));
await settle();
expect(subscriber.tracks$.value).toHaveLength(1);
transceiver.currentDirection = "inactive";
subscriber.handle(offer(current(), { user: { purpose: "m.usermedia" } }));
await settle();
expect(subscriber.tracks$.value).toEqual([]);
});
it("ignores messages of another connection", async () => {
subscriber.handle(offer("other", {}));
await settle();
expect(lastPeerConnection().remoteDescription).toBeNull();
});
it("subscribes anew after the publisher closes, backing off", async () => {
const first = current();
subscriber.handle(
fromPeer(first, { type: SignallingEventType.Close, reason: "nope" }),
);
expect(subscriber.current$.value).toBeNull();
expect(subscriber.tracks$.value).toBeNull();
// It doesn't echo the publisher's close
expect(signalling.sent(SignallingEventType.Close)).toEqual([]);
await vi.advanceTimersByTimeAsync(1_000);
const second = current();
expect(second).not.toBe(first);
expect(signalling.sentConnectionIds(SignallingEventType.Subscribe)).toEqual(
[first, second],
);
// A failed connection is closed and retried after twice the wait
lastPeerConnection().setConnectionState("failed");
expect(signalling.sent(SignallingEventType.Close)).toEqual([
{ type: SignallingEventType.Close, reason: "ice_failed" },
]);
await vi.advanceTimersByTimeAsync(1_000);
expect(signalling.sent(SignallingEventType.Subscribe)).toHaveLength(2);
await vi.advanceTimersByTimeAsync(1_000);
expect(signalling.sent(SignallingEventType.Subscribe)).toHaveLength(3);
});
it("stops retrying once its scope ends", async () => {
subscriber.handle(
fromPeer(current(), { type: SignallingEventType.Close, reason: "nope" }),
);
scope.end();
await vi.advanceTimersByTimeAsync(60_000);
expect(signalling.sent(SignallingEventType.Subscribe)).toHaveLength(1);
});
});
@@ -0,0 +1,297 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
BehaviorSubject,
distinctUntilChanged,
filter,
map,
of,
switchMap,
} from "rxjs";
import { v4 as uuidv4 } from "uuid";
import {
type AudioMediaTrack,
type MediaSource,
type VideoMediaTrack,
} from "../../media-api";
import { type Behavior } from "../../reactive/Behavior";
import { ObservableScope } from "../../reactive/ObservableScope";
import { MediaConnectionState } from "../api";
import { createFullMeshMediaTrack, rtpStats } from "./MediaTrack";
import {
createPeerConnectionBase,
type PeerConnectionBase,
} from "./PeerConnection";
import {
type Peer,
type Signalling,
SignallingEventType,
type SignallingMessage,
type StreamMetadata,
} from "./signalling";
type RemoteTrack = AudioMediaTrack | VideoMediaTrack;
const RETRY_MIN_MS = 1_000;
const RETRY_MAX_MS = 30_000;
export interface SubscriberConnection {
/** The current attempt; null while waiting to retry. */
current$: Behavior<PeerConnectionBase | null>;
/** The publisher's tracks; null until its first offer is applied. */
tracks$: Behavior<RemoteTrack[] | null>;
/** A message from the publisher for the current attempt. */
handle(message: SignallingMessage): void;
}
/**
* Our subscription to one publisher: asks it for its media and answers its
* offers. A failed or closed attempt is retried with a fresh connection id,
* backing off, for as long as `scope` lives.
*/
export function createSubscriberConnection({
scope,
peer,
signalling,
iceServers,
audioOutputDeviceId$,
logger: parentLogger,
}: {
scope: ObservableScope;
peer: Peer;
signalling: Signalling;
iceServers: () => Promise<RTCIceServer[]>;
audioOutputDeviceId$: Behavior<string | undefined>;
logger: Logger;
}): SubscriberConnection {
const logger = parentLogger.getChild(`[Subscriber ← ${peer.memberId}]`);
const attempt$ = new BehaviorSubject<Attempt | null>(null);
let retryMs = RETRY_MIN_MS;
let retryTimer: ReturnType<typeof setTimeout> | undefined;
let ended = false;
scope.onEnd(() => (ended = true));
const startAttempt = async (): Promise<void> => {
const servers = await iceServers();
if (ended) return;
const attemptScope = new ObservableScope();
const attempt = createAttempt({
scope: attemptScope,
peer,
signalling,
iceServers: servers,
audioOutputDeviceId$,
onFailed: (reason) => {
logger.info(`Attempt ${attempt.connectionId} ended: ${reason}`);
attemptScope.end();
if (attempt$.value === attempt) attempt$.next(null);
if (ended) return;
retryTimer = setTimeout(() => void startAttempt(), retryMs);
retryMs = Math.min(retryMs * 2, RETRY_MAX_MS);
},
onConnected: () => (retryMs = RETRY_MIN_MS),
logger,
});
attempt$.next(attempt);
};
void startAttempt();
scope.onEnd(() => {
clearTimeout(retryTimer);
attempt$.value?.end();
attempt$.next(null);
});
return {
current$: scope.behavior(attempt$.pipe(map((a) => a?.base ?? null))),
tracks$: scope.behavior(
attempt$.pipe(switchMap((a) => a?.tracks$ ?? of(null))),
),
handle: (message) => attempt$.value?.handle(message),
};
}
interface Attempt {
connectionId: string;
base: PeerConnectionBase;
tracks$: Behavior<RemoteTrack[] | null>;
handle(message: SignallingMessage): void;
end(): void;
}
function createAttempt({
scope,
peer,
signalling,
iceServers,
audioOutputDeviceId$,
onFailed,
onConnected,
logger,
}: {
scope: ObservableScope;
peer: Peer;
signalling: Signalling;
iceServers: RTCIceServer[];
audioOutputDeviceId$: Behavior<string | undefined>;
onFailed: (reason: string) => void;
onConnected: () => void;
logger: Logger;
}): Attempt {
const connectionId = uuidv4();
let failed = false;
const fail = (reason: string): void => {
if (failed) return;
failed = true;
onFailed(reason);
};
const base = createPeerConnectionBase({
scope,
peer,
connectionId,
signalling,
iceServers,
logger,
});
const { pc, enqueue } = base;
const metadata$ = new BehaviorSubject<StreamMetadata>({});
const tracks$ = new BehaviorSubject<RemoteTrack[] | null>(null);
// The stream each received track arrived in, which the metadata is keyed by
const streamOf = new Map<MediaStreamTrack, string>();
const wrappers = new Map<
RTCRtpTransceiver,
{ track: RemoteTrack; scope: ObservableScope; source: MediaSource }
>();
pc.addEventListener("track", ({ track, streams }) => {
if (streams[0]) streamOf.set(track, streams[0].id);
});
/** The tracks of the transceivers that receive, after each negotiation. */
const updateTracks = (): void => {
const metadata = metadata$.value;
const live = new Set<RTCRtpTransceiver>();
for (const transceiver of pc.getTransceivers()) {
const track = transceiver.receiver.track;
const streamId = streamOf.get(track);
const entry = streamId === undefined ? undefined : metadata[streamId];
if (
transceiver.currentDirection !== "recvonly" ||
entry === undefined ||
(entry.purpose !== "m.usermedia" && entry.purpose !== "m.screenshare")
)
continue;
const kind = track.kind === "audio" ? "audio" : "video";
const source = sourceOf(entry.purpose, kind);
live.add(transceiver);
const existing = wrappers.get(transceiver);
if (existing?.source === source) continue;
existing?.scope.end();
const trackScope = new ObservableScope();
wrappers.set(transceiver, {
scope: trackScope,
source,
track: createFullMeshMediaTrack(trackScope, {
source,
kind,
id: `${connectionId}:${transceiver.mid}`,
track$: trackScope.behavior(of(track)),
muted$: trackScope.behavior(
metadata$.pipe(
map((m) => {
const e = streamId === undefined ? undefined : m[streamId];
return (
(kind === "audio" ? e?.audio_muted : e?.video_muted) ?? false
);
}),
distinctUntilChanged(),
),
),
stats: async () => rtpStats(pc, track, "inbound-rtp"),
audioOutputDeviceId$,
}),
});
}
for (const [transceiver, { scope: s }] of wrappers)
if (!live.has(transceiver)) {
s.end();
wrappers.delete(transceiver);
}
tracks$.next([...wrappers.values()].map(({ track }) => track));
};
base.state$
.pipe(
filter((s) => s instanceof Error || s === MediaConnectionState.Connected),
scope.bind(),
)
.subscribe((s) => {
if (s instanceof Error) {
base.close("ice_failed", true);
fail(s.message);
} else onConnected();
});
const handle = (message: SignallingMessage): void => {
if (message.connectionId !== connectionId) return;
switch (message.type) {
case SignallingEventType.Description:
if (message.description.type !== "offer") return;
enqueue(async () => {
if (message.stream_metadata) metadata$.next(message.stream_metadata);
await pc.setRemoteDescription(message.description);
await pc.setLocalDescription(await pc.createAnswer());
signalling.send(peer, connectionId, {
type: SignallingEventType.Description,
description: { type: "answer", sdp: pc.localDescription!.sdp },
});
updateTracks();
});
break;
case SignallingEventType.Candidates:
base.addCandidates(message);
break;
case SignallingEventType.StreamMetadata:
metadata$.next(message.stream_metadata);
// A stream may only become known with its metadata
enqueue(async () => Promise.resolve(updateTracks()));
break;
case SignallingEventType.Close:
base.close(message.reason, false);
fail(`closed by the publisher: ${message.reason}`);
break;
default:
logger.debug(`Ignoring ${message.type}`);
}
};
scope.onEnd(() => {
for (const { scope: s } of wrappers.values()) s.end();
wrappers.clear();
});
signalling.send(peer, connectionId, { type: SignallingEventType.Subscribe });
return {
connectionId,
base,
tracks$: scope.behavior(tracks$),
handle,
end: () => scope.end(),
};
}
function sourceOf(
purpose: "m.usermedia" | "m.screenshare",
kind: "audio" | "video",
): MediaSource {
if (purpose === "m.usermedia")
return kind === "audio" ? "microphone" : "camera";
return kind === "audio" ? "screenShareAudio" : "screenShare";
}
@@ -0,0 +1,152 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { EventEmitter } from "events";
import { ClientEvent, EventType, MatrixEvent } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import { ObservableScope } from "../../reactive/ObservableScope";
import {
createSignalling,
SEQ_GAP_TIMEOUT_MS,
SeqOrderer,
SignallingEventType,
type SignallingClient,
type SignallingMessage,
} from "./signalling";
describe("SeqOrderer", () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());
it("delivers in seq order", () => {
const delivered: string[] = [];
const orderer = new SeqOrderer<string>((s) => delivered.push(s));
orderer.push(1, "b");
orderer.push(2, "c");
expect(delivered).toEqual([]);
orderer.push(0, "a");
expect(delivered).toEqual(["a", "b", "c"]);
});
it("skips a gap that isn't filled in time, then drops the late message", () => {
const delivered: string[] = [];
const orderer = new SeqOrderer<string>((s) => delivered.push(s));
orderer.push(0, "a");
orderer.push(2, "c");
vi.advanceTimersByTime(SEQ_GAP_TIMEOUT_MS);
expect(delivered).toEqual(["a", "c"]);
orderer.push(1, "b");
expect(delivered).toEqual(["a", "c"]);
});
});
describe("createSignalling", () => {
const own = { userId: "@me:x", deviceId: "MINE", memberId: "me" };
function setup(): {
client: SignallingClient & EventEmitter;
received: SignallingMessage[];
scope: ObservableScope;
} {
const client = Object.assign(new EventEmitter(), {
encryptAndSendToDevice: vi.fn(async () => Promise.resolve()),
}) as unknown as SignallingClient & EventEmitter;
const scope = new ObservableScope();
const signalling = createSignalling({
scope,
client,
roomId: "!room:x",
slotId: "m.call#ROOM",
own,
logger,
});
const received: SignallingMessage[] = [];
signalling.messages$.subscribe((m) => received.push(m));
return { client, received, scope };
}
function toDevice(
type: string,
content: Record<string, unknown>,
encrypted = true,
): MatrixEvent {
const event = new MatrixEvent({ type, sender: "@peer:x", content });
if (encrypted)
vi.spyOn(event, "getWireType").mockReturnValue(
EventType.RoomMessageEncrypted,
);
return event;
}
const common = {
room_id: "!room:x",
slot_id: "m.call#ROOM",
member_id: "peer",
target_member_id: "me",
connection_id: "c1",
device_id: "PEER",
};
it("passes valid messages on and drops cleartext or misaddressed ones", () => {
const { client, received, scope } = setup();
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Subscribe, { ...common, seq: 0 }, false),
);
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Subscribe, {
...common,
target_member_id: "someone else",
seq: 0,
}),
);
expect(received).toEqual([]);
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Subscribe, { ...common, seq: 0 }),
);
expect(received).toEqual([
{
type: SignallingEventType.Subscribe,
memberId: "peer",
userId: "@peer:x",
deviceId: "PEER",
connectionId: "c1",
},
]);
scope.end();
});
it("orders by seq across connection ids", () => {
const { client, received, scope } = setup();
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Subscribe, {
...common,
connection_id: "c2",
seq: 2,
}),
);
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Close, { ...common, seq: 1 }),
);
client.emit(
ClientEvent.ToDeviceEvent,
toDevice(SignallingEventType.Subscribe, { ...common, seq: 0 }),
);
expect(received.map((m) => [m.type, m.connectionId])).toEqual([
[SignallingEventType.Subscribe, "c1"],
[SignallingEventType.Close, "c1"],
[SignallingEventType.Subscribe, "c2"],
]);
scope.end();
});
});
@@ -0,0 +1,304 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
/**
* The to-device signalling of the full mesh transport (MSCXXXX): five event
* types, the fields every one carries, and the ordering by `seq`.
*/
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
ClientEvent,
EventType,
type MatrixClient,
type MatrixEvent,
} from "matrix-js-sdk";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { fromEvent, Observable, share } from "rxjs";
import { type ObservableScope } from "../../reactive/ObservableScope";
export const FULL_MESH_TRANSPORT_TYPE = "org.matrix.mscXXXX.full_mesh";
const prefix = "org.matrix.mscXXXX.rtc.full_mesh";
export const SignallingEventType = {
Subscribe: `${prefix}.subscribe`,
Description: `${prefix}.description`,
Candidates: `${prefix}.candidates`,
StreamMetadata: `${prefix}.stream_metadata`,
Close: `${prefix}.close`,
} as const;
export type SignallingEventType =
(typeof SignallingEventType)[keyof typeof SignallingEventType];
export const signallingEventTypes: string[] =
Object.values(SignallingEventType);
/** How long a gap in `seq` is waited for before the missing messages are skipped. */
export const SEQ_GAP_TIMEOUT_MS = 10_000;
export interface StreamMetadataEntry {
purpose: "m.usermedia" | "m.screenshare";
audio_muted?: boolean;
video_muted?: boolean;
}
/** By `MediaStream` id, as it appears in the SDP's `msid`. */
export type StreamMetadata = Record<string, StreamMetadataEntry>;
interface CommonContent {
room_id: string;
slot_id: string;
member_id: string;
target_member_id: string;
connection_id: string;
seq: number;
/**
* Not in the MSC, which takes the device from the Olm session. The js-sdk
* does not expose that for to-device events, so, like the key transport's
* `claimed_device_id`, the sender states it.
*/
device_id: string;
}
export type SignallingPayload =
| { type: typeof SignallingEventType.Subscribe }
| {
type: typeof SignallingEventType.Description;
description: { type: "offer" | "answer"; sdp: string };
stream_metadata?: StreamMetadata;
}
| {
type: typeof SignallingEventType.Candidates;
candidates: RTCIceCandidateInit[];
}
| {
type: typeof SignallingEventType.StreamMetadata;
stream_metadata: StreamMetadata;
}
| { type: typeof SignallingEventType.Close; reason?: string };
export type SignallingMessage = SignallingPayload & {
/** The member that sent it, as named in `member_id`. */
memberId: string;
userId: string;
deviceId: string;
connectionId: string;
};
/** Another member, as far as signalling needs to know it. */
export interface Peer {
userId: string;
deviceId: string;
memberId: string;
}
export interface Signalling {
send(peer: Peer, connectionId: string, payload: SignallingPayload): void;
/** Every valid message for us, per sender in `seq` order. */
messages$: Observable<SignallingMessage>;
}
export type SignallingClient = Pick<
MatrixClient,
"encryptAndSendToDevice" | "on" | "off"
>;
interface Props {
scope: ObservableScope;
client: SignallingClient;
roomId: string;
slotId: string;
own: CallMembershipIdentityParts;
logger: Logger;
}
export function createSignalling({
scope,
client,
roomId,
slotId,
own,
logger: parentLogger,
}: Props): Signalling {
const logger = parentLogger.getChild("[Signalling]");
const nextSeqTo = new Map<string, number>();
const send = (
peer: Peer,
connectionId: string,
{ type, ...payload }: SignallingPayload,
): void => {
const seq = nextSeqTo.get(peer.memberId) ?? 0;
nextSeqTo.set(peer.memberId, seq + 1);
const content: CommonContent & Record<string, unknown> = {
...payload,
room_id: roomId,
slot_id: slotId,
member_id: own.memberId,
target_member_id: peer.memberId,
connection_id: connectionId,
seq,
device_id: own.deviceId,
};
client
.encryptAndSendToDevice(
type,
[{ userId: peer.userId, deviceId: peer.deviceId }],
content,
)
.catch((e: unknown) => logger.warn(`Failed to send ${type}`, e));
};
const messages$ = new Observable<SignallingMessage>((subscriber) => {
const orderers = new Map<string, SeqOrderer<SignallingMessage>>();
const sub = fromEvent<MatrixEvent>(client, ClientEvent.ToDeviceEvent)
.pipe(scope.bind())
.subscribe((event) => {
const parsed = parse(event, roomId, slotId, own.memberId, logger);
if (parsed === null) return;
let orderer = orderers.get(parsed.message.memberId);
if (orderer === undefined) {
orderer = new SeqOrderer((m) => subscriber.next(m), logger);
orderers.set(parsed.message.memberId, orderer);
}
orderer.push(parsed.seq, parsed.message);
});
return () => {
sub.unsubscribe();
for (const orderer of orderers.values()) orderer.dispose();
};
}).pipe(share());
return { send, messages$ };
}
function parse(
event: MatrixEvent,
roomId: string,
slotId: string,
ownMemberId: string,
logger: Logger,
): { seq: number; message: SignallingMessage } | null {
const type = event.getType();
if (!signallingEventTypes.includes(type)) return null;
// Via the widget API, to-device events are reconstructed as encrypted
if (event.getWireType() !== EventType.RoomMessageEncrypted) {
logger.warn(`Dropping ${type} sent in clear from ${event.getSender()}`);
return null;
}
const c = event.getContent<
Partial<CommonContent> & Record<string, unknown>
>();
const userId = event.getSender();
if (
userId === undefined ||
c.room_id !== roomId ||
c.slot_id !== slotId ||
c.target_member_id !== ownMemberId ||
typeof c.member_id !== "string" ||
typeof c.connection_id !== "string" ||
typeof c.device_id !== "string" ||
typeof c.seq !== "number"
)
return null;
const common = {
memberId: c.member_id,
userId,
deviceId: c.device_id,
connectionId: c.connection_id,
};
let payload: SignallingPayload;
switch (type) {
case SignallingEventType.Subscribe:
payload = { type };
break;
case SignallingEventType.Description: {
const d = c.description as { type?: unknown; sdp?: unknown } | undefined;
if (
(d?.type !== "offer" && d?.type !== "answer") ||
typeof d.sdp !== "string"
)
return null;
payload = {
type,
description: { type: d.type, sdp: d.sdp },
stream_metadata: c.stream_metadata as StreamMetadata | undefined,
};
break;
}
case SignallingEventType.Candidates:
if (!Array.isArray(c.candidates)) return null;
payload = { type, candidates: c.candidates as RTCIceCandidateInit[] };
break;
case SignallingEventType.StreamMetadata:
if (typeof c.stream_metadata !== "object" || c.stream_metadata === null)
return null;
payload = { type, stream_metadata: c.stream_metadata as StreamMetadata };
break;
case SignallingEventType.Close:
payload = {
type,
reason: typeof c.reason === "string" ? c.reason : undefined,
};
break;
default:
return null;
}
return { seq: c.seq, message: { ...payload, ...common } };
}
/**
* Delivers the messages of one sender in `seq` order. A gap is waited for
* `SEQ_GAP_TIMEOUT_MS`, then skipped. A message below the next expected `seq`
* came after its gap was skipped, and is dropped.
*/
export class SeqOrderer<T> {
private next = 0;
private readonly buffer = new Map<number, T>();
private gapTimer: ReturnType<typeof setTimeout> | undefined;
public constructor(
private readonly deliver: (item: T) => void,
private readonly logger?: Logger,
private readonly gapTimeoutMs = SEQ_GAP_TIMEOUT_MS,
) {}
public push(seq: number, item: T): void {
if (seq < this.next) {
this.logger?.warn(`Dropping seq ${seq}, already past it`);
return;
}
this.buffer.set(seq, item);
this.flush();
}
public dispose(): void {
clearTimeout(this.gapTimer);
this.buffer.clear();
}
private flush(): void {
let item = this.buffer.get(this.next);
while (item !== undefined) {
this.buffer.delete(this.next);
this.next++;
this.deliver(item);
item = this.buffer.get(this.next);
}
clearTimeout(this.gapTimer);
this.gapTimer = undefined;
if (this.buffer.size > 0)
this.gapTimer = setTimeout(() => this.skipGap(), this.gapTimeoutMs);
}
private skipGap(): void {
const lowest = Math.min(...this.buffer.keys());
this.logger?.warn(`Skipping seq ${this.next} to ${lowest - 1}`);
this.next = lowest;
this.flush();
}
}
+286
View File
@@ -0,0 +1,286 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
/**
* Fakes for the parts of WebRTC that jsdom lacks, and for the signalling, so
* that the full mesh connections can run in unit tests.
*/
import { vi } from "vitest";
import { BehaviorSubject, Subject } from "rxjs";
import { type LocalPublication } from "./LocalMedia";
import {
type Peer,
type Signalling,
type SignallingMessage,
type SignallingPayload,
type StreamMetadata,
} from "./signalling";
let nextId = 0;
export class FakeMediaStream {
public readonly id: string;
public constructor(public readonly tracks: MediaStreamTrack[] = []) {
this.id = `stream-${nextId++}`;
}
public getTracks(): MediaStreamTrack[] {
return this.tracks;
}
}
export function fakeTrack(kind: "audio" | "video"): MediaStreamTrack {
return {
id: `track-${nextId++}`,
kind,
enabled: true,
stop: vi.fn(),
getSettings: () => ({}),
} as unknown as MediaStreamTrack;
}
interface FakeTransceiver {
kind: string;
mid: string;
direction: RTCRtpTransceiverDirection;
currentDirection: RTCRtpTransceiverDirection | null;
streams: MediaStream[];
stopped: boolean;
stop(): void;
sender: {
track: MediaStreamTrack | null;
replaceTrack: ReturnType<typeof vi.fn>;
};
receiver: { track: MediaStreamTrack };
}
/** Enough of `RTCPeerConnection` for one side of a connection, driven by the test. */
export class FakePeerConnection extends EventTarget {
public static instances: FakePeerConnection[] = [];
public signalingState: RTCSignalingState = "stable";
public connectionState: RTCPeerConnectionState = "new";
public iceConnectionState: RTCIceConnectionState = "new";
public localDescription: RTCSessionDescriptionInit | null = null;
public remoteDescription: RTCSessionDescriptionInit | null = null;
public readonly transceivers: FakeTransceiver[] = [];
public readonly addedCandidates: RTCIceCandidateInit[] = [];
public offers = 0;
public closed = false;
public constructor(public readonly config: RTCConfiguration) {
super();
FakePeerConnection.instances.push(this);
}
public addTransceiver(
kind: string,
init: RTCRtpTransceiverInit,
): FakeTransceiver {
const transceiver: FakeTransceiver = {
kind,
mid: String(this.transceivers.length),
direction: init.direction ?? "sendrecv",
currentDirection: null,
streams: init.streams ?? [],
stopped: false,
stop(): void {
this.stopped = true;
},
sender: {
track: null,
replaceTrack: vi.fn(async (track: MediaStreamTrack | null) => {
transceiver.sender.track = track;
return Promise.resolve();
}),
},
receiver: { track: fakeTrack(kind === "audio" ? "audio" : "video") },
};
this.transceivers.push(transceiver);
return transceiver;
}
public getTransceivers(): FakeTransceiver[] {
return this.transceivers;
}
public getSenders(): FakeTransceiver["sender"][] {
return this.transceivers.map((t) => t.sender);
}
public async createOffer(): Promise<RTCSessionDescriptionInit> {
return Promise.resolve({ type: "offer", sdp: `offer-${++this.offers}` });
}
public async createAnswer(): Promise<RTCSessionDescriptionInit> {
return Promise.resolve({ type: "answer", sdp: "answer" });
}
public async setLocalDescription(
description: RTCSessionDescriptionInit,
): Promise<void> {
this.localDescription = description;
this.signalingState =
description.type === "offer" ? "have-local-offer" : "stable";
return Promise.resolve();
}
public async setRemoteDescription(
description: RTCSessionDescriptionInit,
): Promise<void> {
this.remoteDescription = description;
this.signalingState =
description.type === "offer" ? "have-remote-offer" : "stable";
return Promise.resolve();
}
public async addIceCandidate(candidate: RTCIceCandidateInit): Promise<void> {
if (candidate.candidate === "bad")
return Promise.reject(new Error("bad candidate"));
this.addedCandidates.push(candidate);
return Promise.resolve();
}
public async getStats(): Promise<RTCStatsReport> {
return Promise.resolve(new Map() as unknown as RTCStatsReport);
}
public close(): void {
this.closed = true;
this.signalingState = "closed";
this.connectionState = "closed";
}
/** Gathers a local candidate; null ends the gathering. */
public gather(candidate: RTCIceCandidateInit | null): void {
this.dispatchEvent(
Object.assign(new Event("icecandidate"), {
candidate: candidate && { toJSON: () => candidate },
}),
);
}
public setConnectionState(state: RTCPeerConnectionState): void {
this.connectionState = state;
this.dispatchEvent(new Event("connectionstatechange"));
}
/** A track the remote side sends, as the next negotiation would bring it. */
public receive(kind: "audio" | "video", streamId: string): FakeTransceiver {
const transceiver = this.addTransceiver(kind, { direction: "recvonly" });
transceiver.currentDirection = "recvonly";
this.dispatchEvent(
Object.assign(new Event("track"), {
track: transceiver.receiver.track,
streams: [{ id: streamId }],
}),
);
return transceiver;
}
}
class FakeAudioContext {
public destination = {};
public createMediaStreamSource(): object {
return { connect: vi.fn(), disconnect: vi.fn() };
}
public createAnalyser(): object {
return {
fftSize: 512,
connect: vi.fn(),
disconnect: vi.fn(),
getFloatTimeDomainData: (samples: Float32Array): void => {
samples.fill(0);
},
};
}
public createGain(): object {
return { gain: { value: 1 }, connect: vi.fn(), disconnect: vi.fn() };
}
}
/** Replaces the WebRTC and Web Audio globals jsdom doesn't have. Undo with `vi.unstubAllGlobals()`. */
export function installWebRtcFakes(): void {
FakePeerConnection.instances = [];
vi.stubGlobal("RTCPeerConnection", FakePeerConnection);
vi.stubGlobal("MediaStream", FakeMediaStream);
vi.stubGlobal("AudioContext", FakeAudioContext);
}
/** The last peer connection created. */
export function lastPeerConnection(): FakePeerConnection {
const pc = FakePeerConnection.instances.at(-1);
if (pc === undefined) throw new Error("No peer connection was created");
return pc;
}
/** Lets every queued signalling operation run. */
export async function settle(): Promise<void> {
for (let i = 0; i < 20; i++) await Promise.resolve();
}
export const peer: Peer = {
userId: "@peer:example.org",
deviceId: "PEERDEVICE",
memberId: "peerMember",
};
export interface FakeSignalling extends Signalling {
send: ReturnType<typeof vi.fn<Signalling["send"]>>;
messages$: Subject<SignallingMessage>;
/** The payloads sent so far, optionally only those of one type. */
sent(type?: string): SignallingPayload[];
/** The connection ids sent with the payloads, in the same order. */
sentConnectionIds(type?: string): string[];
}
export function fakeSignalling(): FakeSignalling {
const send = vi.fn<Signalling["send"]>();
const calls = (type?: string): Parameters<Signalling["send"]>[] =>
send.mock.calls.filter(([, , p]) => type === undefined || p.type === type);
return {
send,
messages$: new Subject(),
sent: (type) => calls(type).map(([, , p]) => p),
sentConnectionIds: (type) => calls(type).map(([, id]) => id),
};
}
/** A message from `peer` on a connection. */
export function fromPeer(
connectionId: string,
payload: SignallingPayload,
): SignallingMessage {
return { ...payload, ...peer, connectionId };
}
/** The local media a publisher connection reads. */
export function fakeLocalMedia(): {
publications$: BehaviorSubject<LocalPublication[]>;
publishing$: BehaviorSubject<boolean>;
streamMetadata$: BehaviorSubject<StreamMetadata>;
} {
return {
publications$: new BehaviorSubject<LocalPublication[]>([]),
publishing$: new BehaviorSubject(true),
streamMetadata$: new BehaviorSubject<StreamMetadata>({}),
};
}
export function fakePublication(
kind: "audio" | "video",
stream = new FakeMediaStream() as unknown as MediaStream,
): LocalPublication & { track$: BehaviorSubject<MediaStreamTrack | null> } {
return {
source: kind === "audio" ? "microphone" : "camera",
kind,
stream,
track$: new BehaviorSubject<MediaStreamTrack | null>(fakeTrack(kind)),
mediaTrack: {} as LocalPublication["mediaTrack"],
};
}
+11 -4
View File
@@ -101,8 +101,11 @@ export class ConnectionMap {
interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
/** The transport we publish on; the one connection with the publisher role. */
localTransport: TransportLocator;
/**
* The transport we publish on; the one connection with the publisher role.
* Null for a backend that only subscribes, while another backend publishes.
*/
localTransport: TransportLocator | null;
/** Every transport some member is on, deduplicated against the local one. */
remoteTransports$: Behavior<Epoch<TransportLocator[]>>;
logger: Logger;
@@ -128,7 +131,11 @@ export function createConnectionMap$({
scope.behavior(
remoteTransports$.pipe(
mapEpoch((remoteTransports) =>
removeDuplicateTransports([localTransport, ...remoteTransports]),
removeDuplicateTransports(
localTransport
? [localTransport, ...remoteTransports]
: remoteTransports,
),
),
tap((transports) =>
logger.debug("localAndRemoteTransports$ = ", transports),
@@ -136,7 +143,7 @@ export function createConnectionMap$({
),
);
const localTransportKey = keyFor(localTransport);
const localTransportKey = localTransport && keyFor(localTransport);
const connections$ = scope.behavior(
localAndRemoteTransports$.pipe(
@@ -115,9 +115,14 @@ export function createLivekitBackend(
);
// A new map, in a scope of its own, whenever the local transport changes:
// the publisher role sits on one connection, and which one is fixed per map
// the publisher role sits on one connection, and which one is fixed per map.
// Without a local transport the map only subscribes, for when another
// backend publishes; boxed so that `mapScoped` builds a map for null too
const boxedLocalTransport$ = scope.behavior(
localTransport$.pipe(map((localTransport) => ({ localTransport }))),
);
const connectionMap$ = scope.behavior<Epoch<ConnectionMap>>(
mapScoped(scope, localTransport$, (mapScope, localTransport) =>
mapScoped(scope, boxedLocalTransport$, (mapScope, { localTransport }) =>
createConnectionMap$({
scope: mapScope,
connectionFactory,
@@ -139,12 +144,15 @@ export function createLivekitBackend(
logger,
});
// Probed once: the answer cannot change within a session
const homeserverSupportsDelegation = supportsDelegation(
`${client.baseUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
"homeserver",
logger,
);
// Probed once, when first needed: the answer cannot change within a
// session, and a backend that only subscribes never needs it
let homeserverProbe: Promise<boolean> | undefined;
const homeserverSupportsDelegation = async (): Promise<boolean> =>
(homeserverProbe ??= supportsDelegation(
`${client.baseUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
"homeserver",
logger,
));
const ownServerName = serverNameOf(ownMembershipIdentity.userId);
@@ -176,7 +184,7 @@ export function createLivekitBackend(
`transport ${flow.serviceUrl}`,
logger,
)
: homeserverSupportsDelegation;
: homeserverSupportsDelegation();
}
localTransport$.next(locator);
// The membership is only sent once the publisher connection holds its
+5 -1
View File
@@ -70,6 +70,7 @@ import { ReactionsOverlay } from "./ReactionsOverlay";
import { CallEventAudioRenderer } from "./CallEventAudioRenderer";
import {
createRTCSlot,
FULL_MESH_TRANSPORT_TYPE,
type RTCParticipationOptions,
type RTCSlotOptions,
constant,
@@ -84,6 +85,7 @@ import {
cameraFramerate,
cameraResolution,
customTransport,
fullMeshPublishing,
echoCancellationSetting,
matrixRTCMode as matrixRTCModeSetting,
noiseSuppressionSetting,
@@ -823,7 +825,9 @@ export function rtcParticipationOptions(
delegatedDelayedLeave: session.delegated_delayed_leave,
},
mediaQuality: config.media_quality,
transport: customTransport.value$.value ?? undefined,
transport: fullMeshPublishing.value$.value
? { type: FULL_MESH_TRANSPORT_TYPE }
: (customTransport.value$.value ?? undefined),
fallbackTransport: config.livekit && {
type: "livekit",
livekit_service_url: config.livekit.livekit_service_url,
+28 -2
View File
@@ -45,6 +45,7 @@ import {
alwaysShowIphoneEarpiece as alwaysShowIphoneEarpieceSetting,
matrixRTCMode as matrixRTCModeSetting,
customTransport as customTransportSetting,
fullMeshPublishing as fullMeshPublishingSetting,
advancedScreenShare as advancedScreenShareSetting,
screenShareResolution as screenShareResolutionSetting,
screenShareFramerate as screenShareFramerateSetting,
@@ -68,6 +69,7 @@ import { Slider } from "../Slider";
import { useUrlParams } from "../UrlParams";
import {
authenticateWithTransport,
FULL_MESH_TRANSPORT_TYPE,
type TransportMetadata,
} from "@element-hq/matrixrtc-sdk";
import { useBehavior } from "../useBehavior";
@@ -175,6 +177,9 @@ export const DeveloperSettingsTab: FC<Props> = ({
const [customTransportUpdateError, setCustomTransportUpdateError] = useState<
string | null
>(null);
const [fullMeshPublishing, setFullMeshPublishing] = useSetting(
fullMeshPublishingSetting,
);
const [customTransport, setCustomTransport] = useSetting(
customTransportSetting,
);
@@ -417,6 +422,17 @@ export const DeveloperSettingsTab: FC<Props> = ({
)}
/>
</FieldRow>
<FieldRow>
<InputField
id="fullMeshPublishing"
type="checkbox"
checked={fullMeshPublishing}
label={t("developer_mode.full_mesh_publishing")}
onChange={(event: ChangeEvent<HTMLInputElement>): void =>
setFullMeshPublishing(event.target.checked)
}
/>
</FieldRow>
<FieldRow>
<InputField
id="debugTileLayout"
@@ -665,11 +681,21 @@ function transportText(transport: UnstableLivekitTransport | null): string {
const TransportBox: FC<TransportInfo> = ({ transport, local }) => {
const { t } = useTranslation();
const resolved = useBehavior(transport.resolved$);
const fullMesh = transport.type === FULL_MESH_TRANSPORT_TYPE;
return (
<div className={styles.livekit_room_box}>
<h4>{t("developer_mode.livekit_sfu", { url: transport.id })}</h4>
<h4>
{fullMesh
? t("developer_mode.full_mesh_transport", {
device: transport.raw.device_id,
})
: t("developer_mode.livekit_sfu", { url: transport.id })}
</h4>
<p>
{t("developer_mode.livekit_server_info")}({local ? "local" : "remote"})
{fullMesh
? t("developer_mode.full_mesh_peer_connections")
: t("developer_mode.livekit_server_info")}
({local ? "local" : "remote"})
</p>
<pre className={styles.pre}>
{resolved === undefined
@@ -51,6 +51,43 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
>
<input
aria-describedby="_r_2_"
id="fullMeshPublishing"
type="checkbox"
/>
<label
for="fullMeshPublishing"
>
<div
class="_checkbox_1bd8c0"
>
<svg
fill="none"
height="24"
stroke="#000"
stroke-linecap="round"
stroke-linejoin="round"
stroke-width="2"
viewBox="0 0 24 24"
width="24"
xmlns="http://www.w3.org/2000/svg"
>
<path
d="m20 6-11 11-5-5"
/>
</svg>
</div>
Publish on full mesh (MSCXXXX) instead of LiveKit
</label>
</div>
</div>
<div
class="_fieldRow_1bd8c0"
>
<div
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_3_"
id="debugTileLayout"
type="checkbox"
/>
@@ -87,7 +124,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_3_"
aria-describedby="_r_4_"
id="showConnectionStats"
type="checkbox"
/>
@@ -124,7 +161,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_4_"
aria-describedby="_r_5_"
id="muteAllAudio"
type="checkbox"
/>
@@ -162,7 +199,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_5_"
aria-describedby="_r_6_"
id="alwaysShowIphoneEarpiece"
type="checkbox"
/>
@@ -199,7 +236,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_6_"
aria-describedby="_r_7_"
id="enableLivekitExtendedLogs"
type="checkbox"
/>
@@ -237,7 +274,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
>
<label
class="_label_1o4d9_60"
for="radix-_r_7_"
for="radix-_r_8_"
>
Custom transport JSON
</label>
@@ -245,9 +282,9 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_controls_17lij_8"
>
<input
aria-describedby="radix-_r_8_"
aria-describedby="radix-_r_9_"
class="_control_d83jn_10"
id="radix-_r_7_"
id="radix-_r_8_"
name="input"
title=""
value=""
@@ -255,7 +292,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
</div>
<span
class="_message_1o4d9_86 _help-message_1o4d9_92"
id="radix-_r_8_"
id="radix-_r_9_"
>
Currently, no overwrite is set. Url from config is used.
</span>
@@ -285,10 +322,10 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_container_1ug7n_10"
>
<input
aria-describedby="radix-_r_a_ radix-_r_c_"
aria-describedby="radix-_r_b_ radix-_r_d_"
checked=""
class="_input_1ug7n_18"
id="radix-_r_9_"
id="radix-_r_a_"
name="_r_0_"
title=""
type="radio"
@@ -304,13 +341,13 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
>
<label
class="_label_1o4d9_60"
for="radix-_r_9_"
for="radix-_r_a_"
>
Compatibility: state events & multi SFU
</label>
<span
class="_message_1o4d9_86 _help-message_1o4d9_92"
id="radix-_r_a_"
id="radix-_r_b_"
>
Compatible with homeservers that do not support sticky events (but all other EC clients are v0.17.0 or later)
</span>
@@ -326,9 +363,9 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_container_1ug7n_10"
>
<input
aria-describedby="radix-_r_a_ radix-_r_c_"
aria-describedby="radix-_r_b_ radix-_r_d_"
class="_input_1ug7n_18"
id="radix-_r_b_"
id="radix-_r_c_"
name="_r_0_"
title=""
type="radio"
@@ -344,13 +381,13 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
>
<label
class="_label_1o4d9_60"
for="radix-_r_b_"
for="radix-_r_c_"
>
Matrix 2.0: sticky events & multi SFU
</label>
<span
class="_message_1o4d9_86 _help-message_1o4d9_92"
id="radix-_r_c_"
id="radix-_r_d_"
>
Compatible only with homservers supporting sticky events and all EC clients v0.17.0 or later
</span>
@@ -415,7 +452,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_i_"
aria-describedby="_r_j_"
id="cameraToggle"
type="checkbox"
/>
@@ -445,7 +482,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
</label>
<p
class="_description_1bd8c0"
id="_r_i_"
id="_r_j_"
>
Configure resolution, framerate, bitrate, and codec for camera video. Changes apply on next call join.
</p>
@@ -467,7 +504,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_j_"
aria-describedby="_r_k_"
id="screenShareToggle"
type="checkbox"
/>
@@ -497,7 +534,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
</label>
<p
class="_description_1bd8c0"
id="_r_j_"
id="_r_k_"
>
Configure resolution, framerate, bitrate, and codec for screen sharing
</p>
@@ -522,7 +559,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_k_"
aria-describedby="_r_l_"
checked=""
id="echoCancellation"
type="checkbox"
@@ -560,7 +597,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_l_"
aria-describedby="_r_m_"
checked=""
id="noiseSuppression"
type="checkbox"
@@ -598,7 +635,7 @@ exports[`DeveloperSettingsTab > renders and matches snapshot 1`] = `
class="_field_1bd8c0 _checkboxField_1bd8c0"
>
<input
aria-describedby="_r_m_"
aria-describedby="_r_n_"
checked=""
id="autoGainControl"
type="checkbox"
+9
View File
@@ -159,6 +159,15 @@ export const customTransport = new Setting<UnstableLivekitTransport | null>(
null,
);
/**
* Publish on the full mesh transport (MSCXXXX) instead of LiveKit. Members on
* either transport are received whatever this says.
*/
export const fullMeshPublishing = new Setting<boolean>(
"full-mesh-publishing",
false,
);
export type VideoCodec = "vp8" | "vp9" | "h264" | "av1";
/**
+1 -1
View File
@@ -109,7 +109,7 @@ describe("widget", () => {
receiveState,
sendToDevice: sendRecvToDevice,
receiveToDevice: sendRecvToDevice,
turnServers: false,
turnServers: true,
rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
+11 -2
View File
@@ -21,8 +21,12 @@ import type { IWidgetApiRequest } from "matrix-widget-api";
import { LazyEventEmitter } from "./LazyEventEmitter";
import { getUrlParams } from "./UrlParams";
import { Config } from "./config/Config";
import { seedSettingsFromConfig } from "./settings/settings";
import {
fullMeshPublishing,
seedSettingsFromConfig,
} from "./settings/settings";
import { ElementCallReactionEventType } from "./reactions";
import { fullMeshSignallingEventTypes } from "@element-hq/matrixrtc-sdk";
// Subset of the actions in element-web
export enum ElementWidgetActions {
@@ -162,6 +166,10 @@ export const initializeWidget = (
EventType.CallSDPStreamMetadataChangedPrefix,
EventType.CallReplaces,
EventType.CallEncryptionKeysPrefix,
// Full mesh signalling (MSCXXXX), only where it is turned on: Element
// Web doesn't auto-approve these for Element Call yet, and would ask
// the user on every call
...(fullMeshPublishing.getValue() ? fullMeshSignallingEventTypes : []),
];
const client = createRoomWidgetClient(
@@ -173,7 +181,8 @@ export const initializeWidget = (
receiveState,
sendToDevice: sendRecvToDevice,
receiveToDevice: sendRecvToDevice,
turnServers: false,
// Full mesh connects through the homeserver's TURN server
turnServers: true,
rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,