Update SDK to be consumed by Element Call's CallViewModel

Update the public API to export the primitives and operators the call
model needs, and change `LocalMediaInputs` from observable flags to
immediate booleans with imperative setters. Add session timings, media
quality and capture settings as configuration options, and allow the
host to override the transport discovery with explicit URLs.
This commit is contained in:
Timo K.
2026-10-08 13:53:56 +02:00
parent 31eafb394b
commit 5c6d2f0891
14 changed files with 621 additions and 218 deletions
+6 -4
View File
@@ -9,8 +9,9 @@ in [`SdkMigration.md`](./SdkMigration.md).
**Status:** first implementation. `createMatrixRTCClient` joins the session, connects to **Status:** first implementation. `createMatrixRTCClient` joins the session, connects to
the transport, publishes the local media and exposes every member's media; the the transport, publishes the local media and exposes every member's media; the
development harness and its e2e tests in `playwright/sdk` drive it. It depends on development harness and its e2e tests in `playwright/sdk` drive it, and Element
nothing in Element Call's `src/`; Element Call does not consume it yet. Call's own `CallViewModel` is built on it. It depends on nothing in Element Call's
`src/`.
## Using it ## Using it
@@ -29,10 +30,11 @@ const rtcClient = createMatrixRTCClient(
client, // a matrix-js-sdk MatrixClient, logged in and syncing client, // a matrix-js-sdk MatrixClient, logged in and syncing
room, // the matrix-js-sdk Room, from client.getRoom() once the join has synced room, // the matrix-js-sdk Room, from client.getRoom() once the join has synced
{ {
microphoneEnabled$: constant(true), microphoneEnabled: true,
cameraEnabled$: constant(true), cameraEnabled: true,
audioInputDeviceId$: constant(undefined), audioInputDeviceId$: constant(undefined),
videoInputDeviceId$: constant(undefined), videoInputDeviceId$: constant(undefined),
audioOutputDeviceId$: constant(undefined),
videoProcessor$: constant(undefined), videoProcessor$: constant(undefined),
}, },
{ {
+62 -29
View File
@@ -185,17 +185,31 @@ export interface MatrixRTCClientOptions {
* (`"audio"` or `"video"`); anything else is dropped until it can. * (`"audio"` or `"video"`); anything else is dropped until it can.
*/ */
applicationData?: Record<string, unknown>; applicationData?: Record<string, unknown>;
/** Session timings the host has configured; the defaults otherwise. */
timings?: Partial<SessionTimings>;
/** Limits on what is published; LiveKit's defaults otherwise. */
mediaQuality?: MediaQuality;
/** How the local tracks are captured and encoded: audio processing, camera and screen share resolution, bitrate, codec. */
capture?: CaptureSettings;
/** Use this transport instead of asking the homeserver. */
transportUrl?: string;
/** Use this transport when the homeserver advertises none. */
fallbackTransportUrl?: string;
} }
/** /**
* What the local member publishes. The SDK publishes what it is told; device * What the local member publishes. The enabled flags are the initial state;
* enumeration, permission prompts and the lobby preview stay with the host. * from then on `setMicrophoneEnabled` and `setCameraEnabled` on the client
* change it, and report what the device allowed. Device enumeration,
* permission prompts and the lobby preview stay with the host.
*/ */
export interface LocalMediaInputs { export interface LocalMediaInputs {
microphoneEnabled$: Behavior<boolean>; microphoneEnabled: boolean;
cameraEnabled$: Behavior<boolean>; cameraEnabled: boolean;
audioInputDeviceId$: Behavior<string | undefined>; audioInputDeviceId$: Behavior<string | undefined>;
videoInputDeviceId$: Behavior<string | undefined>; videoInputDeviceId$: Behavior<string | undefined>;
/** Undefined where the host routes audio itself, or to leave the browser's choice. */
audioOutputDeviceId$: Behavior<string | undefined>;
/** Background blur and the like. */ /** Background blur and the like. */
videoProcessor$: Behavior<TrackProcessor<Track.Kind.Video> | undefined>; videoProcessor$: Behavior<TrackProcessor<Track.Kind.Video> | undefined>;
} }
@@ -230,8 +244,17 @@ export type ConnectionStatus =
| "reconnecting" | "reconnecting"
| "disconnected"; | "disconnected";
/** An error raised by the client. `cause` holds the backend's error, for a host that knows the backend. */ /** Why the client is not connected: the first failing of its three links. */
export class MatrixRTCError extends Error {} export type DisconnectReason = "sync" | "membership" | "probablyLeft" | "media";
/**
* An error raised by the client. `code` and `category` say what went wrong in
* a form a host can translate; `cause` holds the backend's error.
*/
export class MatrixRTCError extends Error {
code: ErrorCode;
category: ErrorCategory;
}
export interface MatrixRTCClient { export interface MatrixRTCClient {
join(): void; join(): void;
@@ -242,6 +265,17 @@ export interface MatrixRTCClient {
connected$: Behavior<boolean>; connected$: Behavior<boolean>;
/** Connected once, and currently not. */ /** Connected once, and currently not. */
reconnecting$: Behavior<boolean>; reconnecting$: Behavior<boolean>;
/** Null while connected, the first failing link otherwise. */
disconnectReason$: Behavior<DisconnectReason | null>;
/**
* Publishes or mutes the microphone. Resolves with the state that
* resulted, which differs from the request where the device could not be
* used. Before the transport is connected the request is remembered and
* applied once it is.
*/
setMicrophoneEnabled(enabled: boolean): Promise<boolean>;
setCameraEnabled(enabled: boolean): Promise<boolean>;
/** /**
* A transport, Matrix or connection error that stops the session. Null * A transport, Matrix or connection error that stops the session. Null
* while fine. A failed publication is not fatal: the member can still * while fine. A failed publication is not fatal: the member can still
@@ -459,6 +493,12 @@ export interface MemberMedia {
export interface LocalMemberMedia extends MemberMedia { export interface LocalMemberMedia extends MemberMedia {
local: true; local: true;
/**
* Restarts the camera facing the other way, on devices with a front and a
* back camera, and resolves with the id of the device now in use. Does
* nothing without a camera track or where the facing mode is unknown.
*/
switchCamera(): Promise<string | undefined>;
} }
``` ```
@@ -549,27 +589,20 @@ public API only.
## Open decisions ## Open decisions
- **Timings.** Delayed-leave timings, the sync grace period, the key rotation - **`LocalMediaInputs` shape.** The enabled flags are initial values and the
delay and participant limit come from `DEFAULT_CONFIG` today. The SDK must not setters on the client change them, because the setters have to report what
read `config.json`, so they become an option (`timings`, with these defaults). the device allowed; the device ids and the processor stay behaviors. Creating
Delegating the delayed leave to the SFU is not implemented, since it needs the the tracks before `join()` for a preview is still missing: today they are
same timings; the js-sdk restarts the delayed event itself meanwhile. created at the join.
- **`LocalMediaInputs` as behaviors or methods.** Behaviors match how the mute - **`applicationData` after the join.** The option is read once. The client
states drive the publisher and keep a lobby's pre-join state in one place; still updates `m.call.intent` from the camera state, as Element Call did; that
methods are the more conventional SDK shape. Two things are missing either way: moves to the host once the data can be updated.
creating the tracks before `join()` for a preview (today they are created at the
join), and screen share capture options, which are fixed.
- **`applicationData` after the join.** The option is read once. The implementation
still updates `m.call.intent` from the camera input, as Element Call does; that
moves to the host once the data can be updated, as a behavior or a setter.
- **Backend overrides.** Tests of the client itself need to inject the transport,
the connection factory and the LiveKit room. They were planned as
`options.backend`, grouped so that no LiveKit type appears on the options; not
yet added, so `createMatrixRTCClient` is covered by the e2e tests only.
- **Raw local state.** The local member's state machine (`LocalMemberState`) stays - **Raw local state.** The local member's state machine (`LocalMemberState`) stays
internal; `status$` and `fatalError$` are its public view. A developer panel may internal; `status$`, `disconnectReason$` and `fatalError$` are its public view.
want it as `LocalRTCMember.state$`. A developer panel may want it as `LocalRTCMember.state$`.
- **Developer panel.** `resolved$.identity` covers the local identity and - **Developer panel.** `resolved$` on each connected transport covers what the
`remoteMembers$` filtered by `transport$.id` the remote list. Anything beyond panel shows today. Anything beyond that (LiveKit room state, connection
that (LiveKit room state, connection quality) needs an opaque `debug$` per quality) needs an opaque `debug$` per transport.
transport. - **Unencrypted rooms with per-participant keys.** Keys only reach devices the
crypto tracks, which are the members of encrypted rooms. If the SDK is to work
in an unencrypted room, tracking the members' devices becomes its job.
+13 -8
View File
@@ -9,21 +9,26 @@ symbols, so it ages with the code; the architecture document should not.
## Status ## Status
| Slice | State | | Slice | State |
| --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- |
| 1. Scaffold: `sdk/` packaging, the interface, the harness, the smoke test | done, October 2026 | | 1. Scaffold: `sdk/` packaging, the interface, the harness, the smoke test | done, October 2026 |
| 2. First implementation of `createMatrixRTCClient` under `sdk/src`, the media adapter, the e2e suite | done, 1 October 2026 | | 2. First implementation of `createMatrixRTCClient` under `sdk/src`, the media adapter, the e2e suite | done, 1 October 2026 |
| 3. Copy the shared modules into `sdk/src/` with their tests, drop every `src/` import, turn on the import boundary | done, 2 October 2026 | | 3. Copy the shared modules into `sdk/src/` with their tests, drop every `src/` import, turn on the import boundary | done, 2 October 2026 |
| 4. `createCallViewModel$` consumes the client; media view models take `media$`; Element Call imports the primitives and enums from the SDK and deletes its own copies | next | | 4. `createCallViewModel$` consumes the client; media view models take `media$`; Element Call imports the primitives and enums from the SDK and deletes its own copies; the old demo target is removed | done, 2 October 2026 |
| 5. Remove `sdk-target-based-on-call-view-model/` and the old `build:sdk` output | last |
`src/` was not touched in slices 2 and 3: the SDK has its own copies of what it Element Call now consumes the SDK. `InCallView` creates the client from the
needs, and the lint rule `element-call/sdk-import-boundary` fails any import from configuration, settings and URL parameters, and hands it to `createCallViewModel$`,
`sdk/` that resolves outside `sdk/` or names React, i18n or Compound. Until slice 4 which keeps what makes the session a call. The modules below were deleted from
the two copies drift independently, which is why slice 4 comes next. Nothing in `src/` once the SDK had its copies, and `Behavior`, `ObservableScope`, the
Element Call consumes the SDK yet. observable operators, `E2eeType`, `EncryptionSystem` and `MatrixRTCMode` are
re-exported from the SDK by their old paths so the rest of Element Call did not
have to change in the same step. The view model tests drive a fake
`MatrixRTCClient` (`mockMatrixRTCClient` in `src/utils/test.ts`) instead of mocked
LiveKit rooms; the tests of the deleted modules live on in `sdk/src`.
## What was copied from `src/`, and what changed on the way ## What was copied from `src/`, and what changed on the way
All of these are gone from `src/` as of slice 4; the table records where they went.
Each module came with its tests (`sdk/src/utils/test.ts` and `test-fixtures.ts` are Each module came with its tests (`sdk/src/utils/test.ts` and `test-fixtures.ts` are
the subset of `src/utils/test.ts` and `test-fixtures.ts` they need). In slice 4 the subset of `src/utils/test.ts` and `test-fixtures.ts` they need). In slice 4
Element Call switches to the SDK's copy and deletes its own. Element Call switches to the SDK's copy and deletes its own.
+15 -16
View File
@@ -12,13 +12,7 @@ Please see LICENSE in the repository root for full details.
*/ */
import { logger } from "matrix-js-sdk/lib/logger"; import { logger } from "matrix-js-sdk/lib/logger";
import { import { combineLatest, type Observable, of, switchMap } from "rxjs";
BehaviorSubject,
combineLatest,
type Observable,
of,
switchMap,
} from "rxjs";
import { import {
constant, constant,
createMatrixRTCClient, createMatrixRTCClient,
@@ -71,17 +65,16 @@ async function start(
const room = await joinRoom(client, roomIdOrAlias); const room = await joinRoom(client, roomIdOrAlias);
const scope = new ObservableScope(); const scope = new ObservableScope();
const microphoneEnabled$ = new BehaviorSubject(true);
const cameraEnabled$ = new BehaviorSubject(true);
const rtcClient = createMatrixRTCClient( const rtcClient = createMatrixRTCClient(
scope, scope,
client, client,
room, room,
{ {
microphoneEnabled$, microphoneEnabled: true,
cameraEnabled$, cameraEnabled: true,
audioInputDeviceId$: constant(undefined), audioInputDeviceId$: constant(undefined),
videoInputDeviceId$: constant(undefined), videoInputDeviceId$: constant(undefined),
audioOutputDeviceId$: constant(undefined),
videoProcessor$: constant(undefined), videoProcessor$: constant(undefined),
}, },
{ {
@@ -98,8 +91,12 @@ async function start(
showMembers(scope, rtcClient); showMembers(scope, rtcClient);
rtcClient.join(); rtcClient.join();
toggle(buttons.microphone, microphoneEnabled$); toggle(buttons.microphone, async (enabled) =>
toggle(buttons.camera, cameraEnabled$); rtcClient.setMicrophoneEnabled(enabled),
);
toggle(buttons.camera, async (enabled) =>
rtcClient.setCameraEnabled(enabled),
);
buttons.leave.hidden = false; buttons.leave.hidden = false;
buttons.leave.onclick = (): void => { buttons.leave.onclick = (): void => {
rtcClient.leave(); rtcClient.leave();
@@ -114,14 +111,16 @@ async function start(
} }
} }
/** A pressed button means the source is on; the SDK says what it could do. */
function toggle( function toggle(
button: HTMLButtonElement, button: HTMLButtonElement,
enabled$: BehaviorSubject<boolean>, set: (enabled: boolean) => Promise<boolean>,
): void { ): void {
button.hidden = false; button.hidden = false;
button.onclick = (): void => { button.onclick = (): void => {
enabled$.next(!enabled$.value); void set(button.ariaPressed !== "true").then((enabled) => {
button.ariaPressed = String(enabled$.value); button.ariaPressed = String(enabled);
});
}; };
} }
+40 -9
View File
@@ -10,18 +10,49 @@ Please see LICENSE in the repository root for full details.
* *
* MatrixRTC sessions with LiveKit media, without Element Call's UI: the call * MatrixRTC sessions with LiveKit media, without Element Call's UI: the call
* model Element Call's own view model is built on, for hosts that want to * model Element Call's own view model is built on, for hosts that want to
* build a different one. The design and the migration from Element Call's * build a different one. The design is in `SdkArchitecture.md` next to this
* `CallViewModel` are in `sdk-plan.md` at the repository root. * file.
*/ */
// Shared with Element Call. They live in `src` until the view model consumes
// the SDK, at which point they move here; a consumer gets them from this
// module either way.
export { type Behavior, constant } from "./src/reactive/Behavior"; export { type Behavior, constant } from "./src/reactive/Behavior";
export { ObservableScope } from "./src/reactive/ObservableScope"; export {
export { E2eeType } from "./src/encryption"; Epoch,
export { type EncryptionSystem } from "./src/encryption"; mapEpoch,
export { MatrixRTCMode } from "./src/config"; ObservableScope,
trackEpoch,
} from "./src/reactive/ObservableScope";
export {
filterBehavior,
finalizeValue,
generateItems,
generateItemsWithEpoch,
pauseWhen,
} from "./src/reactive/observable";
export { E2eeType, type EncryptionSystem } from "./src/encryption";
export {
defaultMediaQuality,
defaultSessionTimings,
type DelayedLeaveTimings,
MatrixRTCMode,
type MediaQuality,
type SessionTimings,
} from "./src/config";
export {
ErrorCategory,
ErrorCode,
FailToGetOpenIdToken,
FailToStartLivekitConnection,
InsufficientCapacityError,
LivekitConnectionError,
MatrixRTCTransportMissingError,
MembershipManagerError,
NoMatrix2AuthorizationService,
PeerConnectionTimeoutError,
SFURoomCreationRestrictedError,
UnknownCallError as UnknownRTCError,
} from "./src/errors";
export * from "./src/api"; export * from "./src/api";
export { createMatrixRTCClient } from "./src/session/MatrixRTCClient"; export { createMatrixRTCClient } from "./src/session/MatrixRTCClient";
// For a developer panel to check a transport url before using it
export { getSFUConfigWithOpenID as authenticateWithTransport } from "./src/session/openIDSFU";
+68 -5
View File
@@ -15,12 +15,20 @@ import {
type RTCNotificationType, type RTCNotificationType,
type Transport, type Transport,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type Track, type TrackProcessor } from "livekit-client"; import {
type Track,
type TrackProcessor,
type VideoCodec,
} from "livekit-client";
import { type Observable } from "rxjs"; import { type Observable } from "rxjs";
import { type Behavior } from "./reactive/Behavior"; import { type Behavior } from "./reactive/Behavior";
import { type EncryptionSystem } from "./encryption"; import { type EncryptionSystem } from "./encryption";
import { type MatrixRTCMode } from "./config"; import {
type MatrixRTCMode,
type MediaQuality,
type SessionTimings,
} from "./config";
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Session // Session
@@ -45,14 +53,48 @@ export interface MatrixRTCClientOptions {
* (`"audio"` or `"video"`); anything else is dropped until it can. * (`"audio"` or `"video"`); anything else is dropped until it can.
*/ */
applicationData?: Record<string, unknown>; applicationData?: Record<string, unknown>;
/** Session timings the host has configured; the defaults otherwise. */
timings?: Partial<SessionTimings>;
/** Limits on what is published; LiveKit's defaults otherwise. */
mediaQuality?: MediaQuality;
/** How the local tracks are captured and encoded. */
capture?: CaptureSettings;
/** Use this transport instead of asking the homeserver. */
transportUrl?: string;
/** Use this transport when the homeserver advertises none. */
fallbackTransportUrl?: string;
} }
/** What the local member publishes. */ /** Capture and encoding settings for the local tracks, applied at creation. */
export interface CaptureSettings {
audio?: {
echoCancellation?: boolean;
noiseSuppression?: boolean;
autoGainControl?: boolean;
};
camera?: VideoCaptureSettings;
screenShare?: VideoCaptureSettings;
}
export interface VideoCaptureSettings {
resolution?: { width: number; height: number; frameRate?: number };
maxBitrate?: number;
maxFramerate?: number;
codec?: VideoCodec;
}
/**
* What the local member publishes. The enabled flags are the initial state;
* from then on `setMicrophoneEnabled` and `setCameraEnabled` on the client
* change it, and report what the device allowed.
*/
export interface LocalMediaInputs { export interface LocalMediaInputs {
microphoneEnabled$: Behavior<boolean>; microphoneEnabled: boolean;
cameraEnabled$: Behavior<boolean>; cameraEnabled: boolean;
audioInputDeviceId$: Behavior<string | undefined>; audioInputDeviceId$: Behavior<string | undefined>;
videoInputDeviceId$: Behavior<string | undefined>; videoInputDeviceId$: Behavior<string | undefined>;
/** Undefined where the host routes audio itself, or to leave the browser's choice. */
audioOutputDeviceId$: Behavior<string | undefined>;
/** Background blur and the like. */ /** Background blur and the like. */
videoProcessor$: Behavior<TrackProcessor<Track.Kind.Video> | undefined>; videoProcessor$: Behavior<TrackProcessor<Track.Kind.Video> | undefined>;
} }
@@ -64,6 +106,9 @@ export type ConnectionStatus =
| "reconnecting" | "reconnecting"
| "disconnected"; | "disconnected";
/** Why the client is not connected: the first failing of its three links. */
export type DisconnectReason = "sync" | "membership" | "probablyLeft" | "media";
import { type MatrixRTCError } from "./errors"; import { type MatrixRTCError } from "./errors";
export { MatrixRTCError } from "./errors"; export { MatrixRTCError } from "./errors";
@@ -81,6 +126,17 @@ export interface MatrixRTCClient {
status$: Behavior<ConnectionStatus>; status$: Behavior<ConnectionStatus>;
connected$: Behavior<boolean>; connected$: Behavior<boolean>;
reconnecting$: Behavior<boolean>; reconnecting$: Behavior<boolean>;
/** Null while connected, the first failing link otherwise. */
disconnectReason$: Behavior<DisconnectReason | null>;
/**
* Publishes or mutes the microphone. Resolves with the state that
* resulted, which differs from the request where the device could not be
* used. Before the transport is connected the request is remembered and
* applied once it is.
*/
setMicrophoneEnabled(enabled: boolean): Promise<boolean>;
setCameraEnabled(enabled: boolean): Promise<boolean>;
/** A transport, Matrix or connection error that stops the session. */ /** A transport, Matrix or connection error that stops the session. */
fatalError$: Behavior<MatrixRTCError | null>; fatalError$: Behavior<MatrixRTCError | null>;
@@ -239,4 +295,11 @@ export interface MemberMedia {
export interface LocalMemberMedia extends MemberMedia { export interface LocalMemberMedia extends MemberMedia {
local: true; local: true;
/**
* Restarts the camera facing the other way, on devices with a front and a
* back camera, and resolves with the id of the device now in use. Does
* nothing without a camera track or where the facing mode is unknown (see
* `VideoMediaTrack.facingMode$`).
*/
switchCamera(): Promise<string | undefined>;
} }
+7 -1
View File
@@ -24,7 +24,7 @@ export interface DelayedLeaveTimings {
/** How long (in milliseconds) the homeserver waits before sending the delayed leave. */ /** How long (in milliseconds) the homeserver waits before sending the delayed leave. */
delay_ms: number; delay_ms: number;
/** How often (in milliseconds) the client restarts the delayed leave. */ /** How often (in milliseconds) the client restarts the delayed leave. */
restart_ms: number; restart_ms?: number;
/** How long (in milliseconds) a restart may take before it counts as failed. */ /** How long (in milliseconds) a restart may take before it counts as failed. */
restart_timeout_ms?: number; restart_timeout_ms?: number;
} }
@@ -45,12 +45,18 @@ export interface SessionTimings {
/** The session size at which the media key stops being rotated. */ /** The session size at which the media key stops being rotated. */
keyRotationParticipantLimit?: number; keyRotationParticipantLimit?: number;
delayedLeave: DelayedLeaveTimings; delayedLeave: DelayedLeaveTimings;
/**
* The timings used where the SFU restarts the delayed leave on the
* client's behalf, which lets the client stop renewing it itself.
*/
delegatedDelayedLeave: DelayedLeaveTimings;
} }
export const defaultSessionTimings: SessionTimings = { export const defaultSessionTimings: SessionTimings = {
syncDisconnectGracePeriodMs: 10_000, syncDisconnectGracePeriodMs: 10_000,
networkErrorRetryMs: 1_000, networkErrorRetryMs: 1_000,
delayedLeave: { delay_ms: 18_000, restart_ms: 4_000 }, delayedLeave: { delay_ms: 18_000, restart_ms: 4_000 },
delegatedDelayedLeave: { delay_ms: 3_600_000, restart_ms: 300_000 },
}; };
/** Media quality settings a host may pass; everything is optional. */ /** Media quality settings a host may pass; everything is optional. */
+13
View File
@@ -11,11 +11,14 @@ import {
roomEventSelector, roomEventSelector,
} from "@livekit/components-core"; } from "@livekit/components-core";
import { import {
facingModeFromLocalTrack,
type LocalParticipant, type LocalParticipant,
LocalVideoTrack,
type Participant, type Participant,
ParticipantEvent, ParticipantEvent,
type Room as LivekitRoom, type Room as LivekitRoom,
RoomEvent, RoomEvent,
Track,
type TrackPublication, type TrackPublication,
} from "livekit-client"; } from "livekit-client";
import { distinctUntilChanged, filter, map, type Observable } from "rxjs"; import { distinctUntilChanged, filter, map, type Observable } from "rxjs";
@@ -89,6 +92,16 @@ export function createLocalLivekitMemberMedia(
return { return {
...createLivekitMemberMedia(scope, participant, room, encryptionSystem), ...createLivekitMemberMedia(scope, participant, room, encryptionSystem),
local: true, local: true,
switchCamera: async () => {
const track = participant.getTrackPublication(Track.Source.Camera)?.track;
if (!(track instanceof LocalVideoTrack)) return;
const { facingMode } = facingModeFromLocalTrack(track);
if (facingMode !== "user" && facingMode !== "environment") return;
await track.restartTrack({
facingMode: facingMode === "user" ? "environment" : "user",
});
return track.mediaStreamTrack.getSettings().deviceId;
},
}; };
} }
+33 -3
View File
@@ -23,7 +23,12 @@ import { type OpenIDClientParts, type SFUConfig } from "./openIDSFU";
import { type Behavior } from "../reactive/Behavior"; import { type Behavior } from "../reactive/Behavior";
import { Connection, type ConnectionOpts, ConnectionState } from "./Connection"; import { Connection, type ConnectionOpts, ConnectionState } from "./Connection";
import { type ObservableScope } from "../reactive/ObservableScope"; import { type ObservableScope } from "../reactive/ObservableScope";
import { type LocalMediaInputs, type ResolvedTransport } from "../api"; import {
type CaptureSettings,
type LocalMediaInputs,
type ResolvedTransport,
} from "../api";
import { type MediaQuality } from "../config";
export interface ConnectionFactory { export interface ConnectionFactory {
createConnection( createConnection(
@@ -87,6 +92,8 @@ export class LivekitConnectionFactory implements ConnectionFactory {
private readonly roomId: string, private readonly roomId: string,
private readonly localMedia: LocalMediaInputs, private readonly localMedia: LocalMediaInputs,
private readonly keyProvider: BaseKeyProvider | undefined, private readonly keyProvider: BaseKeyProvider | undefined,
private readonly mediaQuality: MediaQuality | undefined,
private readonly capture: CaptureSettings | undefined,
) {} ) {}
public createConnection( public createConnection(
@@ -104,7 +111,14 @@ export class LivekitConnectionFactory implements ConnectionFactory {
client: this.client, client: this.client,
scope, scope,
livekitRoomFactory: () => livekitRoomFactory: () =>
new LivekitRoom(roomOptions(this.localMedia, this.keyProvider)), new LivekitRoom(
roomOptions(
this.localMedia,
this.keyProvider,
this.mediaQuality,
this.capture,
),
),
ownMembershipIdentity, ownMembershipIdentity,
}, },
logger, logger,
@@ -115,19 +129,35 @@ export class LivekitConnectionFactory implements ConnectionFactory {
function roomOptions( function roomOptions(
localMedia: LocalMediaInputs, localMedia: LocalMediaInputs,
keyProvider: BaseKeyProvider | undefined, keyProvider: BaseKeyProvider | undefined,
mediaQuality: MediaQuality | undefined,
capture: CaptureSettings | undefined,
): RoomOptions { ): RoomOptions {
const base = buildLiveKitOptions(); const base = buildLiveKitOptions(mediaQuality);
const camera = capture?.camera;
return { return {
...base, ...base,
videoCaptureDefaults: { videoCaptureDefaults: {
...base.videoCaptureDefaults, ...base.videoCaptureDefaults,
deviceId: localMedia.videoInputDeviceId$.value, deviceId: localMedia.videoInputDeviceId$.value,
processor: localMedia.videoProcessor$.value, processor: localMedia.videoProcessor$.value,
...(camera?.resolution && { resolution: camera.resolution }),
},
publishDefaults: {
...base.publishDefaults,
...(camera?.maxBitrate !== undefined && {
videoEncoding: {
maxBitrate: camera.maxBitrate,
maxFramerate: camera.maxFramerate,
},
}),
...(camera?.codec && { videoCodec: camera.codec }),
}, },
audioCaptureDefaults: { audioCaptureDefaults: {
...base.audioCaptureDefaults, ...base.audioCaptureDefaults,
deviceId: localMedia.audioInputDeviceId$.value, deviceId: localMedia.audioInputDeviceId$.value,
...capture?.audio,
}, },
audioOutput: { deviceId: localMedia.audioOutputDeviceId$.value },
// Every room needs a worker of its own: one gets confused by streams from // Every room needs a worker of its own: one gets confused by streams from
// several rooms // several rooms
e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() }, e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() },
+190 -38
View File
@@ -11,13 +11,17 @@ import {
type Participant, type Participant,
ParticipantEvent, ParticipantEvent,
type ScreenShareCaptureOptions, type ScreenShareCaptureOptions,
Track,
type TrackPublishOptions,
} from "livekit-client"; } from "livekit-client";
import { type MatrixClient } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { import {
type LivekitTransport, type LivekitTransport,
type MatrixRTCSession as JsSdkRTCSession, type MatrixRTCSession as JsSdkRTCSession,
type Status as RTCSessionStatus, type Status as RTCSessionStatus,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { deepCompare } from "matrix-js-sdk/lib/utils"; import { deepCompare } from "matrix-js-sdk/lib/utils";
import { import {
BehaviorSubject, BehaviorSubject,
@@ -37,17 +41,24 @@ import {
import { type Behavior } from "../reactive/Behavior"; import { type Behavior } from "../reactive/Behavior";
import { type ObservableScope } from "../reactive/ObservableScope"; import { type ObservableScope } from "../reactive/ObservableScope";
import { type HomeserverConnected } from "./HomeserverConnected"; import {
import { type Connection, ConnectionState } from "./Connection"; type DelayedLeaveTimings,
import { type IConnectionManager } from "./ConnectionManager"; type MatrixRTCMode,
type SessionTimings,
} from "../config";
import { type DisconnectReason, type VideoCaptureSettings } from "../api";
import { import {
FailToStartLivekitConnection, FailToStartLivekitConnection,
type MatrixRTCError, type MatrixRTCError,
MembershipManagerError, MembershipManagerError,
toMatrixRTCError, toMatrixRTCError,
} from "../errors"; } from "../errors";
import { type HomeserverConnected } from "./HomeserverConnected";
import { type Connection, ConnectionState } from "./Connection";
import { type IConnectionManager } from "./ConnectionManager";
import { type LocalTransport } from "./LocalTransport"; import { type LocalTransport } from "./LocalTransport";
import { type Publisher } from "./Publisher"; import { getSFUConfigWithOpenID } from "./openIDSFU";
import { type DesiredMedia, type Publisher } from "./Publisher";
export enum TransportState { export enum TransportState {
Waiting = "transport_waiting", Waiting = "transport_waiting",
@@ -66,10 +77,7 @@ export type LocalMemberMediaState =
export type LocalMemberState = export type LocalMemberState =
| MatrixRTCError | MatrixRTCError
| TransportState.Waiting | TransportState.Waiting
| { | { media: LocalMemberMediaState; matrix: MatrixRTCError | RTCSessionStatus };
media: LocalMemberMediaState;
matrix: MatrixRTCError | RTCSessionStatus;
};
interface Props { interface Props {
scope: ObservableScope; scope: ObservableScope;
@@ -77,14 +85,22 @@ interface Props {
localTransport$: Observable<LocalTransport>; localTransport$: Observable<LocalTransport>;
homeserverConnected: HomeserverConnected; homeserverConnected: HomeserverConnected;
createPublisher: (connection: Connection) => Publisher; createPublisher: (connection: Connection) => Publisher;
joinMatrixRTC: (transport: LivekitTransport) => void; joinMatrixRTC: (
transport: LivekitTransport,
delayedLeave: DelayedLeaveTimings,
) => void;
/** The membership manager giving up on keeping the membership alive. */ /** The membership manager giving up on keeping the membership alive. */
membershipManagerError$: Observable<unknown>; membershipManagerError$: Observable<unknown>;
matrixRTCSession: Pick< matrixRTCSession: Pick<JsSdkRTCSession, "leaveRoomSession">;
JsSdkRTCSession, /** The id of the delayed leave event, for the SFU to take over. */
"updateCallIntent" | "leaveRoomSession" delayId$: Behavior<string | null>;
>; client: Pick<MatrixClient, "getOpenIdToken" | "getDeviceId" | "baseUrl">;
cameraEnabled$: Behavior<boolean>; roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
matrixRTCMode: MatrixRTCMode;
timings: SessionTimings;
desired: DesiredMedia;
screenShare?: VideoCaptureSettings;
logger: Logger; logger: Logger;
} }
@@ -99,6 +115,9 @@ export interface LocalMembership {
connected$: Behavior<boolean>; connected$: Behavior<boolean>;
/** Connected once, and currently not. */ /** Connected once, and currently not. */
reconnecting$: Behavior<boolean>; reconnecting$: Behavior<boolean>;
disconnectReason$: Behavior<DisconnectReason | null>;
setMicrophoneEnabled: (enabled: boolean) => Promise<boolean>;
setCameraEnabled: (enabled: boolean) => Promise<boolean>;
sharingScreen$: Behavior<boolean>; sharingScreen$: Behavior<boolean>;
toggleScreenSharing: (() => void) | null; toggleScreenSharing: (() => void) | null;
screenShareError$: Behavior<Error | null>; screenShareError$: Behavior<Error | null>;
@@ -119,7 +138,14 @@ export function createLocalMembership$({
joinMatrixRTC, joinMatrixRTC,
membershipManagerError$, membershipManagerError$,
matrixRTCSession, matrixRTCSession,
cameraEnabled$, delayId$,
client,
roomId,
ownMembershipIdentity,
matrixRTCMode,
timings,
desired,
screenShare,
logger: parentLogger, logger: parentLogger,
}: Props): LocalMembership { }: Props): LocalMembership {
const logger = parentLogger.getChild("[LocalMember]"); const logger = parentLogger.getChild("[LocalMember]");
@@ -131,8 +157,28 @@ export function createLocalMembership$({
return NEVER; return NEVER;
}), }),
); );
const transport$ = scope.behavior<LocalTransport | null>(
localTransport$, // Whether the SFU can take over restarting the delayed leave, so that a
// client that vanishes is removed by the SFU rather than by the timeout.
// Either the homeserver or the transport has to support it.
const homeserverSupportsDelegation = checkDelegationSupport(
`${client.baseUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
"homeserver",
logger,
);
const joinParams$ = scope.behavior(
localTransport$.pipe(
switchMap(async ({ transport }) => ({
transport,
delegationSupported:
(await homeserverSupportsDelegation) ||
(await checkDelegationSupport(
`${transport.livekit_service_url}/delegate_delayed_leave`,
`transport ${transport.livekit_service_url}`,
logger,
)),
})),
),
null, null,
); );
@@ -185,11 +231,16 @@ export function createLocalMembership$({
); );
scope.reconcile( scope.reconcile(
scope.behavior(combineLatest([transport$, joinRequested$])), scope.behavior(combineLatest([joinParams$, joinRequested$])),
async ([transport, shouldJoin]) => { async ([joinParams, shouldJoin]) => {
if (transport === null || !shouldJoin) return; if (joinParams === null || !shouldJoin) return;
try { try {
joinMatrixRTC(transport.transport); joinMatrixRTC(
joinParams.transport,
joinParams.delegationSupported
? timings.delegatedDelayedLeave
: timings.delayedLeave,
);
} catch (e) { } catch (e) {
logger.error("Failed to enter the session", e); logger.error("Failed to enter the session", e);
if (matrixError$.value === null) matrixError$.next(toMatrixRTCError(e)); if (matrixError$.value === null) matrixError$.next(toMatrixRTCError(e));
@@ -204,6 +255,35 @@ export function createLocalMembership$({
}, },
); );
// Hand the delayed leave to the SFU. The token this issues is discarded;
// the request is what triggers the delegation.
scope.reconcile(
scope.behavior(combineLatest([joinParams$, delayId$])),
async ([joinParams, delayId]) => {
if (!joinParams?.delegationSupported || delayId === null) return;
try {
await getSFUConfigWithOpenID(
client,
ownMembershipIdentity,
joinParams.transport.livekit_service_url,
roomId,
{
matrixRTCMode,
delayEndpointBaseUrl: client.baseUrl,
delayId,
delayTimeoutMs: timings.delegatedDelayedLeave.delay_ms,
},
logger,
);
} catch (e) {
logger.error(
`Failed to delegate the leave to ${joinParams.transport.livekit_service_url}`,
e,
);
}
},
);
membershipManagerError$.pipe(scope.bind()).subscribe((e) => { membershipManagerError$.pipe(scope.bind()).subscribe((e) => {
logger.error("The membership manager stopped", e); logger.error("The membership manager stopped", e);
if (matrixError$.value === null) if (matrixError$.value === null)
@@ -214,16 +294,6 @@ export function createLocalMembership$({
); );
}); });
cameraEnabled$.pipe(scope.bind()).subscribe((videoEnabled) => {
matrixRTCSession
.updateCallIntent(videoEnabled ? "video" : "audio")
.catch((e) => {
// Expected before the join: the intent is sent with it instead
if (e instanceof Error && e.message === "Not connected yet") return;
logger.error("Failed to update the call intent", e);
});
});
const connectionState$ = connection$.pipe( const connectionState$ = connection$.pipe(
switchMap((connection) => connection?.state$ ?? of(null)), switchMap((connection) => connection?.state$ ?? of(null)),
); );
@@ -271,15 +341,25 @@ export function createLocalMembership$({
), ),
); );
const connected$ = scope.behavior( const disconnectReason$ = scope.behavior(
combineLatest( combineLatest(
[homeserverConnected.combined$, connectionState$], [homeserverConnected.combined$, connectionState$],
([homeserverConnected], connectionState) => (
homeserverConnected && [homeserverConnected, reason],
connectionState === ConnectionState.LivekitConnected, connectionState,
): DisconnectReason | null => {
if (!homeserverConnected) return reason ?? "sync";
if (connectionState !== ConnectionState.LivekitConnected)
return "media";
return null;
},
), ),
); );
const connected$ = scope.behavior(
disconnectReason$.pipe(map((reason) => reason === null)),
);
const reconnecting$ = scope.behavior( const reconnecting$ = scope.behavior(
connected$.pipe( connected$.pipe(
pairwise(), pairwise(),
@@ -311,6 +391,23 @@ export function createLocalMembership$({
} }
}); });
const setEnabled = async (
source: Track.Source.Microphone | Track.Source.Camera,
enabled: boolean,
): Promise<boolean> => {
const desired$ =
source === Track.Source.Microphone
? desired.microphone$
: desired.camera$;
desired$.next(enabled);
// Without a publisher the request waits for the tracks to be created
const publisher = publisher$.value;
if (publisher === null) return enabled;
const result = await publisher.setEnabled(source, enabled);
if (result !== enabled) desired$.next(result);
return result;
};
const sharingScreen$ = scope.behavior( const sharingScreen$ = scope.behavior(
participant$.pipe( participant$.pipe(
switchMap((p) => (p === null ? of(false) : observeSharingScreen$(p))), switchMap((p) => (p === null ? of(false) : observeSharingScreen$(p))),
@@ -324,7 +421,11 @@ export function createLocalMembership$({
if (participant === null) return; if (participant === null) return;
const enable = !sharingScreen$.value; const enable = !sharingScreen$.value;
participant participant
.setScreenShareEnabled(enable, screenShareCaptureOptions) .setScreenShareEnabled(
enable,
screenShareCaptureOptions(screenShare),
screenSharePublishOptions(screenShare),
)
.catch((e: unknown) => { .catch((e: unknown) => {
logger.error( logger.error(
`Screen share ${enable ? "start" : "stop"} failed`, `Screen share ${enable ? "start" : "stop"} failed`,
@@ -349,6 +450,11 @@ export function createLocalMembership$({
connection$, connection$,
connected$, connected$,
reconnecting$, reconnecting$,
disconnectReason$,
setMicrophoneEnabled: async (enabled) =>
setEnabled(Track.Source.Microphone, enabled),
setCameraEnabled: async (enabled) =>
setEnabled(Track.Source.Camera, enabled),
sharingScreen$, sharingScreen$,
toggleScreenSharing, toggleScreenSharing,
screenShareError$, screenShareError$,
@@ -356,7 +462,36 @@ export function createLocalMembership$({
}; };
} }
const screenShareCaptureOptions: ScreenShareCaptureOptions = { /**
* Whether an endpoint exists, by hitting it without credentials and reading
* the status. Not retried: many servers predate the endpoint altogether and
* answer with a CORS failure that a retry loop would only repeat.
*/
async function checkDelegationSupport(
endpointUrl: string,
serviceName: string,
logger: Logger,
): Promise<boolean> {
try {
const res = await fetch(endpointUrl, { method: "POST" });
const supported = res.status !== 404;
logger.info(
`${serviceName} ${supported ? "supports" : "does not support"} delegation`,
);
return supported;
} catch (e) {
logger.warn(
`Failed to determine whether ${serviceName} supports delegation, assuming no support`,
e,
);
return false;
}
}
function screenShareCaptureOptions(
settings: VideoCaptureSettings | undefined,
): ScreenShareCaptureOptions {
return {
// No echo cancellation: it would cancel the other members' voices out of // No echo cancellation: it would cancel the other members' voices out of
// the shared audio // the shared audio
audio: { audio: {
@@ -367,7 +502,24 @@ const screenShareCaptureOptions: ScreenShareCaptureOptions = {
selfBrowserSurface: "include", selfBrowserSurface: "include",
surfaceSwitching: "include", surfaceSwitching: "include",
systemAudio: "include", systemAudio: "include",
}; ...(settings?.resolution && { resolution: settings.resolution }),
};
}
function screenSharePublishOptions(
settings: VideoCaptureSettings | undefined,
): TrackPublishOptions | undefined {
if (settings === undefined) return undefined;
return {
...(settings.maxBitrate !== undefined && {
screenShareEncoding: {
maxBitrate: settings.maxBitrate,
maxFramerate: settings.maxFramerate,
},
}),
...(settings.codec && { videoCodec: settings.codec }),
};
}
function observeSharingScreen$(participant: Participant): Observable<boolean> { function observeSharingScreen$(participant: Participant): Observable<boolean> {
return observeParticipantEvents( return observeParticipantEvents(
+13 -5
View File
@@ -38,24 +38,32 @@ interface Props {
ownMembershipIdentity: CallMembershipIdentityParts; ownMembershipIdentity: CallMembershipIdentityParts;
roomId: string; roomId: string;
matrixRTCMode: MatrixRTCMode; matrixRTCMode: MatrixRTCMode;
/** Skip discovery and use this transport. */
transportUrl?: string;
/** Use this transport when the homeserver advertises none. */
fallbackTransportUrl?: string;
logger: Logger; logger: Logger;
} }
/** /**
* The transport the local member publishes on: the homeserver's preferred * The transport the local member publishes on: the one the host names, else
* one, authenticated with so that the session can be joined with a token in * the homeserver's preferred one, else the host's fallback; authenticated
* hand. Only the homeserver is asked; the SDK has no configuration of its own * with, so that the session can be joined with a token in hand.
* to fall back on.
*/ */
export async function getLocalTransport({ export async function getLocalTransport({
client, client,
ownMembershipIdentity, ownMembershipIdentity,
roomId, roomId,
matrixRTCMode, matrixRTCMode,
transportUrl,
fallbackTransportUrl,
logger, logger,
}: Props): Promise<LocalTransport> { }: Props): Promise<LocalTransport> {
const transport = await new RtcTransportAutoDiscovery({ const transport: LivekitTransport | null = transportUrl
? { type: "livekit", livekit_service_url: transportUrl }
: await new RtcTransportAutoDiscovery({
client, client,
fallbackTransportUrl,
logger, logger,
}).discoverPreferredTransport(); }).discoverPreferredTransport();
if (transport === null) if (transport === null)
+69 -28
View File
@@ -10,46 +10,55 @@ import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { import {
MatrixRTCSessionEvent, MatrixRTCSessionEvent,
MatrixRTCSessionManager, MatrixRTCSessionManager,
MembershipManagerEvent,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { v4 as uuidv4 } from "uuid"; import { v4 as uuidv4 } from "uuid";
import { combineLatest, from, fromEvent, map } from "rxjs";
import { MatrixRTCMode } from "../config";
import { type ObservableScope } from "../reactive/ObservableScope";
import { import {
createKeyRotationSuppressed$, BehaviorSubject,
createMemberships$, combineLatest,
membershipsAndTransports$, from,
} from "./SessionBehaviors"; fromEvent,
map,
type Observable,
} from "rxjs";
import { defaultSessionTimings, MatrixRTCMode } from "../config";
import { type ObservableScope } from "../reactive/ObservableScope";
import { filterBehavior, generateItems } from "../reactive/observable";
import { mapScoped } from "../utils/mapScoped";
import {
type LocalMediaInputs,
type LocalRTCMember,
type MatrixRTCClient,
type MatrixRTCClientOptions,
type RemoteRTCMember,
} from "../api";
import { MatrixRTCError } from "../errors";
import { LivekitConnectionFactory } from "./ConnectionFactory";
import { createHomeserverConnected$ } from "./HomeserverConnected"; import { createHomeserverConnected$ } from "./HomeserverConnected";
import { createConnectionManager$ } from "./ConnectionManager"; import { joinJsSdkSession } from "./joinJsSdkSession";
import { createKeyProvider } from "./KeyProvider";
import { createLocalMembership$ } from "./LocalMember";
import { getLocalTransport } from "./LocalTransport";
import { createRemoteMatrixLivekitMembers$ } from "./MatrixLivekitMembers"; import { createRemoteMatrixLivekitMembers$ } from "./MatrixLivekitMembers";
import { import {
createMatrixMemberMetadata$, createMatrixMemberMetadata$,
createRoomMembers$, createRoomMembers$,
} from "./MatrixMemberMetadata"; } from "./MatrixMemberMetadata";
import { filterBehavior, generateItems } from "../reactive/observable";
import {
type LocalMediaInputs,
type LocalRTCMember,
type RemoteRTCMember,
type MatrixRTCClient,
MatrixRTCError,
type MatrixRTCClientOptions,
} from "../api";
import { LivekitConnectionFactory } from "./ConnectionFactory";
import { joinJsSdkSession, sessionTimings } from "./joinJsSdkSession";
import { createKeyProvider } from "./KeyProvider";
import { createLocalMembership$ } from "./LocalMember";
import { getLocalTransport } from "./LocalTransport";
import { import {
createLocalRTCMember, createLocalRTCMember,
createRemoteRTCMember, createRemoteRTCMember,
membershipKeys, membershipKeys,
} from "./Members"; } from "./Members";
import { Publisher } from "./Publisher"; import { createConnectionManager$ } from "./ConnectionManager";
import { mapScoped } from "../utils/mapScoped"; import { type DesiredMedia, Publisher } from "./Publisher";
import {
createKeyRotationSuppressed$,
createMemberships$,
membershipsAndTransports$,
} from "./SessionBehaviors";
import { fatalError, sessionStatus } from "./status"; import { fatalError, sessionStatus } from "./status";
import { createTransportRegistry } from "./Transports"; import { createTransportRegistry } from "./Transports";
@@ -73,6 +82,7 @@ export function createMatrixRTCClient(
if (!(userId && deviceId)) if (!(userId && deviceId))
throw new MatrixRTCError("The client has to be logged in"); throw new MatrixRTCError("The client has to be logged in");
const { encryptionSystem, matrixRTCMode } = options; const { encryptionSystem, matrixRTCMode } = options;
const timings = { ...defaultSessionTimings, ...options.timings };
const jsSdkSession = sessionManager( const jsSdkSession = sessionManager(
scope, scope,
@@ -106,10 +116,17 @@ export function createMatrixRTCClient(
ownMembershipIdentity, ownMembershipIdentity,
roomId: room.roomId, roomId: room.roomId,
matrixRTCMode, matrixRTCMode,
transportUrl: options.transportUrl,
fallbackTransportUrl: options.fallbackTransportUrl,
logger, logger,
}), }),
); );
const desired: DesiredMedia = {
microphone$: new BehaviorSubject(localMedia.microphoneEnabled),
camera$: new BehaviorSubject(localMedia.cameraEnabled),
};
const connectionManager = createConnectionManager$({ const connectionManager = createConnectionManager$({
scope, scope,
connectionFactory: new LivekitConnectionFactory( connectionFactory: new LivekitConnectionFactory(
@@ -117,6 +134,8 @@ export function createMatrixRTCClient(
room.roomId, room.roomId,
localMedia, localMedia,
keyProvider, keyProvider,
options.mediaQuality,
options.capture,
), ),
localTransport$, localTransport$,
remoteTransports$: transports$, remoteTransports$: transports$,
@@ -133,29 +152,48 @@ export function createMatrixRTCClient(
scope, scope,
client, client,
jsSdkSession, jsSdkSession,
sessionTimings.syncDisconnectGracePeriodMs, timings.syncDisconnectGracePeriodMs,
), ),
createPublisher: (connection) => createPublisher: (connection) =>
new Publisher( new Publisher(
connection.livekitRoom, connection.livekitRoom,
localMedia, localMedia,
desired,
logger.getChild( logger.getChild(
`[Publisher ${connection.transport.livekit_service_url}]`, `[Publisher ${connection.transport.livekit_service_url}]`,
), ),
), ),
joinMatrixRTC: (transport) => joinMatrixRTC: (transport, delayedLeave) =>
joinJsSdkSession(jsSdkSession, ownMembershipIdentity, transport, { joinJsSdkSession(jsSdkSession, ownMembershipIdentity, transport, {
encryptMedia: keyProvider !== undefined, encryptMedia: keyProvider !== undefined,
matrixRTCMode, matrixRTCMode,
sendNotificationType: options.sendNotificationType, sendNotificationType: options.sendNotificationType,
applicationData: options.applicationData, applicationData: options.applicationData,
timings,
delayedLeave,
}), }),
membershipManagerError$: fromEvent( membershipManagerError$: fromEvent(
jsSdkSession, jsSdkSession,
MatrixRTCSessionEvent.MembershipManagerError, MatrixRTCSessionEvent.MembershipManagerError,
), ),
matrixRTCSession: jsSdkSession, matrixRTCSession: jsSdkSession,
cameraEnabled$: localMedia.cameraEnabled$, delayId$: scope.behavior(
(
fromEvent(
jsSdkSession,
MembershipManagerEvent.DelayIdChanged,
// The re-emitted event carries the original emitter as the second argument
) as Observable<[string | undefined, IMembershipManager]>
).pipe(map(([delayId]) => delayId ?? null)),
jsSdkSession.delayId ?? null,
),
client,
roomId: room.roomId,
ownMembershipIdentity,
matrixRTCMode,
timings,
desired,
screenShare: options.capture?.screenShare,
logger, logger,
}); });
@@ -228,6 +266,9 @@ export function createMatrixRTCClient(
status$, status$,
connected$: localMembership.connected$, connected$: localMembership.connected$,
reconnecting$: localMembership.reconnecting$, reconnecting$: localMembership.reconnecting$,
disconnectReason$: localMembership.disconnectReason$,
setMicrophoneEnabled: localMembership.setMicrophoneEnabled,
setCameraEnabled: localMembership.setCameraEnabled,
fatalError$: scope.behavior(localMembership.state$.pipe(map(fatalError))), fatalError$: scope.behavior(localMembership.state$.pipe(map(fatalError))),
localMember$, localMember$,
remoteMembers$, remoteMembers$,
+55 -40
View File
@@ -15,14 +15,26 @@ import {
Track, Track,
} from "livekit-client"; } from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { combineLatest, distinctUntilChanged, map, skip } from "rxjs"; import {
type BehaviorSubject,
combineLatest,
distinctUntilChanged,
map,
} from "rxjs";
import { type Behavior } from "../reactive/Behavior"; import { type Behavior } from "../reactive/Behavior";
import { ObservableScope } from "../reactive/ObservableScope"; import { ObservableScope } from "../reactive/ObservableScope";
import { type LocalMediaInputs } from "../api"; import { type LocalMediaInputs } from "../api";
/** What the host last asked for, kept across publishers. */
export interface DesiredMedia {
microphone$: BehaviorSubject<boolean>;
camera$: BehaviorSubject<boolean>;
}
/** /**
* Publishes the local media on one LiveKit room, following `LocalMediaInputs`. * Publishes the local media on one LiveKit room, following the host's
* requests.
* *
* LiveKit publishes a track the moment it is created, but a member must not be * LiveKit publishes a track the moment it is created, but a member must not be
* heard before it has joined the MatrixRTC session, nor after it has left. The * heard before it has joined the MatrixRTC session, nor after it has left. The
@@ -38,6 +50,7 @@ export class Publisher {
public constructor( public constructor(
room: LivekitRoom, room: LivekitRoom,
private readonly inputs: LocalMediaInputs, private readonly inputs: LocalMediaInputs,
private readonly desired: DesiredMedia,
private readonly logger: Logger, private readonly logger: Logger,
) { ) {
this.room = room; this.room = room;
@@ -67,24 +80,47 @@ export class Publisher {
} }
/** /**
* Creates the microphone and camera tracks the inputs ask for, and keeps * Creates the microphone and camera tracks the host asked for. Both are
* them in step with the inputs from then on. Both are enabled in one call so * enabled in one call so that the browser asks for permission once. Safe to
* that the browser asks for permission once. Safe to call more than once. * call more than once.
*/ */
public createAndSetupTracks(): void { public createAndSetupTracks(): void {
if (this.tracksRequested) return; if (this.tracksRequested) return;
this.tracksRequested = true; this.tracksRequested = true;
const participant = this.room.localParticipant; const participant = this.room.localParticipant;
const audio = this.inputs.microphoneEnabled$.value; const audio = this.desired.microphone$.value;
const video = this.inputs.cameraEnabled$.value; const video = this.desired.camera$.value;
// LiveKit resolves these once the track is published, which may block on // LiveKit resolves these once the track is published, which may block on
// the connection; LocalTrackPublished is what tells us a track exists. // the connection; LocalTrackPublished is what tells us a track exists.
if (audio && video) void participant.enableCameraAndMicrophone(); if (audio && video) void participant.enableCameraAndMicrophone();
else if (audio) void participant.setMicrophoneEnabled(true); else if (audio) void participant.setMicrophoneEnabled(true);
else if (video) void participant.setCameraEnabled(true); else if (video) void participant.setCameraEnabled(true);
}
this.follow(this.inputs.microphoneEnabled$, Track.Source.Microphone); /**
this.follow(this.inputs.cameraEnabled$, Track.Source.Camera); * Enables or disables a source and reports what resulted: the request where
* it succeeded, the state LiveKit is left in where it did not (a denied
* permission, a missing device).
*/
public async setEnabled(
source: Track.Source.Microphone | Track.Source.Camera,
enabled: boolean,
): Promise<boolean> {
const participant = this.room.localParticipant;
try {
if (source === Track.Source.Microphone)
await participant.setMicrophoneEnabled(enabled);
else await participant.setCameraEnabled(enabled);
// Unmuting restarts the upstream; until the member has joined, it has
// to stay paused
if (enabled && !this.shouldPublish) await this.pauseUpstreams([source]);
return enabled;
} catch (e) {
this.logger.error(`Failed to set ${source} enabled=${enabled}`, e);
return source === Track.Source.Microphone
? participant.isMicrophoneEnabled
: participant.isCameraEnabled;
}
} }
public async startPublishing(): Promise<void> { public async startPublishing(): Promise<void> {
@@ -122,40 +158,18 @@ export class Publisher {
this.pauseUpstreams([publication.source]).catch((e) => { this.pauseUpstreams([publication.source]).catch((e) => {
this.logger.error("Failed to pause the upstream", e); this.logger.error("Failed to pause the upstream", e);
}); });
// The input may have changed while the track was being created // The host may have changed its mind while the track was being created
const enabled = const desired =
publication.source === Track.Source.Microphone publication.source === Track.Source.Microphone
? this.inputs.microphoneEnabled$.value ? this.desired.microphone$.value
: publication.source === Track.Source.Camera : publication.source === Track.Source.Camera
? this.inputs.cameraEnabled$.value ? this.desired.camera$.value
: undefined; : undefined;
if (enabled === false) this.setEnabled(publication.source, false); if (desired === false)
} void this.setEnabled(
publication.source as Track.Source.Microphone | Track.Source.Camera,
private follow( false,
enabled$: Behavior<boolean>, );
source: Track.Source.Microphone | Track.Source.Camera,
): void {
enabled$
.pipe(skip(1), distinctUntilChanged(), this.scope.bind())
.subscribe((enabled) => this.setEnabled(source, enabled));
}
private setEnabled(source: Track.Source, enabled: boolean): void {
const participant = this.room.localParticipant;
const toggle =
source === Track.Source.Microphone
? participant.setMicrophoneEnabled(enabled)
: participant.setCameraEnabled(enabled);
toggle
.then(async () => {
// Unmuting restarts the upstream; until the member has joined, it
// has to stay paused
if (enabled && !this.shouldPublish) await this.pauseUpstreams([source]);
})
.catch((e) => {
this.logger.error(`Failed to set ${source} enabled=${enabled}`, e);
});
} }
private async pauseUpstreams(sources: Track.Source[]): Promise<void> { private async pauseUpstreams(sources: Track.Source[]): Promise<void> {
@@ -192,6 +206,7 @@ export class Publisher {
}); });
}; };
sync("audioinput", this.inputs.audioInputDeviceId$); sync("audioinput", this.inputs.audioInputDeviceId$);
sync("audiooutput", this.inputs.audioOutputDeviceId$);
sync("videoinput", this.inputs.videoInputDeviceId$); sync("videoinput", this.inputs.videoInputDeviceId$);
} }
+20 -15
View File
@@ -13,18 +13,22 @@ import {
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { defaultSessionTimings, MatrixRTCMode } from "../config"; import {
type DelayedLeaveTimings,
MatrixRTCMode,
type SessionTimings,
} from "../config";
interface Options { interface Options {
encryptMedia: boolean; encryptMedia: boolean;
matrixRTCMode: MatrixRTCMode; matrixRTCMode: MatrixRTCMode;
sendNotificationType?: RTCNotificationType; sendNotificationType?: RTCNotificationType;
applicationData?: Record<string, unknown>; applicationData?: Record<string, unknown>;
timings: SessionTimings;
/** The delayed leave timings to use: delegated to the SFU or not. */
delayedLeave: DelayedLeaveTimings;
} }
/** The session's timings; a later slice makes them an input. */
export const sessionTimings = defaultSessionTimings;
/** /**
* Sends the membership and starts the membership manager, which keeps it * Sends the membership and starts the membership manager, which keeps it
* alive and retries on failure. * alive and retries on failure.
@@ -39,33 +43,34 @@ export function joinJsSdkSession(
matrixRTCMode, matrixRTCMode,
sendNotificationType, sendNotificationType,
applicationData, applicationData,
timings,
delayedLeave,
}: Options, }: Options,
): void { ): void {
// The one piece of application data the js-sdk can put in a membership // The one piece of application data the js-sdk can put in a membership
const intent = applicationData?.["m.call.intent"]; const intent = applicationData?.["m.call.intent"];
const callIntent = const callIntent =
typeof intent === "string" ? (intent as RTCCallIntent) : undefined; typeof intent === "string" ? (intent as RTCCallIntent) : undefined;
const timings = sessionTimings.delayedLeave;
// Give up on the network as soon as either the sync has been down for the // Give up on the network as soon as either the sync has been down for the
// grace period or the delayed leave has probably been sent, whichever is // grace period or the delayed leave has probably been sent, whichever is
// sooner // sooner
const maxWaitMs = Math.min( const maxWaitMs = Math.min(
sessionTimings.syncDisconnectGracePeriodMs, timings.syncDisconnectGracePeriodMs,
timings.delay_ms, delayedLeave.delay_ms,
); );
session.joinRTCSession(ownMembershipIdentity, [transport], { session.joinRTCSession(ownMembershipIdentity, [transport], {
notificationType: sendNotificationType, notificationType: sendNotificationType,
callIntent, callIntent,
manageMediaKeys: encryptMedia, manageMediaKeys: encryptMedia,
delayedLeaveEventRestartMs: timings.restart_ms, delayedLeaveEventRestartMs: delayedLeave.restart_ms,
delayedLeaveEventDelayMs: timings.delay_ms, delayedLeaveEventDelayMs: delayedLeave.delay_ms,
delayedLeaveEventRestartLocalTimeoutMs: timings.restart_timeout_ms, delayedLeaveEventRestartLocalTimeoutMs: delayedLeave.restart_timeout_ms,
networkErrorRetryMs: sessionTimings.networkErrorRetryMs, networkErrorRetryMs: timings.networkErrorRetryMs,
makeKeyDelay: sessionTimings.waitForKeyRotationMs, makeKeyDelay: timings.waitForKeyRotationMs,
membershipEventExpiryMs: sessionTimings.membershipEventExpiryMs, membershipEventExpiryMs: timings.membershipEventExpiryMs,
keyRotationParticipantLimit: sessionTimings.keyRotationParticipantLimit, keyRotationParticipantLimit: timings.keyRotationParticipantLimit,
unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0, unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount:
Math.ceil(maxWaitMs / sessionTimings.networkErrorRetryMs) + 1, Math.ceil(maxWaitMs / timings.networkErrorRetryMs) + 1,
}); });
} }