Publish and unpublish on the local member

One way to put media on the session: publish(request) resolves with the local track once LiveKit has it, and unpublish(id) removes it. A request names a source with its device, processor and capture settings, so CaptureSettings on the client options goes, as do setMicrophoneEnabled and setCameraEnabled on the client and the four screen share fields on the local member; sharing is tracks$ having a screenShare track, and a capture failure rejects the promise. LocalMediaInputs becomes the list of initial requests, which also become the room defaults so that one permission prompt covers microphone and camera. The publisher keeps the requests across a reconnection and republishes what is not muted. The harness gains a Share screen button.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Timo K.
2026-10-08 13:53:57 +02:00
co-authored by Claude Fable 5.1
parent 5a43eaf5a6
commit b9f79be106
12 changed files with 441 additions and 326 deletions
+1 -2
View File
@@ -30,8 +30,7 @@ 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: true, publish: [{ source: "microphone" }, { source: "camera" }],
cameraEnabled: true,
audioOutputDeviceId$: constant(undefined), audioOutputDeviceId$: constant(undefined),
}, },
{ {
+40 -32
View File
@@ -189,8 +189,6 @@ export interface MatrixRTCClientOptions {
timings?: Partial<SessionTimings>; timings?: Partial<SessionTimings>;
/** Limits on what is published; LiveKit's defaults otherwise. */ /** Limits on what is published; LiveKit's defaults otherwise. */
mediaQuality?: MediaQuality; 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. */ /** Use this transport instead of asking the homeserver. */
transportUrl?: string; transportUrl?: string;
/** Use this transport when the homeserver advertises none. */ /** Use this transport when the homeserver advertises none. */
@@ -198,18 +196,33 @@ export interface MatrixRTCClientOptions {
} }
/** /**
* What the local member publishes when it joins. Initial values: from then on * One thing to publish: a source, where to capture it from and how to encode
* `setMicrophoneEnabled` and `setCameraEnabled` on the client, and the * it. Where the device and the settings are left out, the browser's and
* controls on the local tracks, change it. Device enumeration, permission * LiveKit's defaults apply.
* prompts and the lobby preview stay with the host. */
export type PublishRequest =
| { source: "microphone"; deviceId?: string; capture?: AudioCaptureSettings }
| {
source: "camera";
deviceId?: string;
/** Background blur and the like. */
processor?: TrackProcessor<Track.Kind.Video>;
capture?: VideoCaptureSettings;
}
| {
source: "screenShare";
/** Whether to capture the screen's audio too. Default true. */
audio?: boolean;
capture?: VideoCaptureSettings;
};
/**
* Device enumeration, permission prompts and the lobby preview stay with the
* host.
*/ */
export interface LocalMediaInputs { export interface LocalMediaInputs {
microphoneEnabled: boolean; /** Published at the join; `publish` on the local member adds to it from then on. */
cameraEnabled: boolean; publish: PublishRequest[];
audioInputDeviceId?: string;
videoInputDeviceId?: string;
/** Background blur and the like. */
videoProcessor?: TrackProcessor<Track.Kind.Video>;
/** Undefined where the host routes audio itself, or to leave the browser's choice. */ /** Undefined where the host routes audio itself, or to leave the browser's choice. */
audioOutputDeviceId$: Behavior<string | undefined>; audioOutputDeviceId$: Behavior<string | undefined>;
} }
@@ -268,14 +281,6 @@ export interface MatrixRTCClient {
/** Null while connected, the first failing link otherwise. */ /** Null while connected, the first failing link otherwise. */
disconnectReason$: Behavior<DisconnectReason | null>; 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
@@ -367,12 +372,18 @@ export interface RemoteRTCMember extends RTCMember {
export interface LocalRTCMember extends RTCMember { export interface LocalRTCMember extends RTCMember {
local: true; local: true;
media$: Behavior<LocalMemberMedia | null>; media$: Behavior<LocalMemberMedia | null>;
sharingScreen$: Behavior<boolean>; /**
/** Null when the platform cannot share a screen (no getDisplayMedia). Hiding the button is the host's policy. */ * Publishes a source and resolves with its track once it is in `tracks$`.
toggleScreenSharing: (() => void) | null; * Before the transport is connected the request is remembered and applied
/** The last error from toggling, until dismissed. The user closing the picker is not one. */ * once it is. Rejects where the device could not be used, including the
screenShareError$: Behavior<Error | null>; * user closing the picker. One publication per source: publishing a source
dismissScreenShareError(): void; * again unmutes it.
*/
publish(
request: PublishRequest,
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
/** Removes one of our tracks; a screen share takes its audio with it. */
unpublish(id: string): Promise<void>;
} }
``` ```
@@ -426,7 +437,7 @@ they only make sense for a call:
| `autoLeave$` / `AutoLeaveReason` (`"allOthersLeft" \| "timeout" \| "decline"`) | hanging up because the other side declined or everyone left is call etiquette, not session semantics | `CallViewModel.leave$`, built from `remoteMembers$` and the room timeline | | `autoLeave$` / `AutoLeaveReason` (`"allOthersLeft" \| "timeout" \| "decline"`) | hanging up because the other side declined or everyone left is call etiquette, not session semantics | `CallViewModel.leave$`, built from `remoteMembers$` and the room timeline |
| `ringAttempts$`, `createCallNotificationLifecycle$`, `createSentCallNotification$`, `createReceivedDecline$` | ringing, pickup timeouts and decline events are MSC4075 call-notify UX | `src/state/CallViewModel/CallNotificationLifecycle.ts`, fed by `remoteMembers$` and the Matrix room | | `ringAttempts$`, `createCallNotificationLifecycle$`, `createSentCallNotification$`, `createReceivedDecline$` | ringing, pickup timeouts and decline events are MSC4075 call-notify UX | `src/state/CallViewModel/CallNotificationLifecycle.ts`, fed by `remoteMembers$` and the Matrix room |
| `waitForCallPickup`, `autoLeaveWhenOthersLeft` options | inputs to the above | `CallViewModelOptions` | | `waitForCallPickup`, `autoLeaveWhenOthersLeft` options | inputs to the above | `CallViewModelOptions` |
| `hideScreensharing` option | a UI policy; the session can always share a screen if the platform can | `CallViewModel` nulls `toggleScreenSharing` | | `hideScreensharing` option | a UI policy; the session can always share a screen if the platform can | `CallViewModel` offers no screen share control |
| `ringingVm$`, `ringingStatusLocation` | view models | `CallViewModel` | | `ringingVm$`, `ringingStatusLocation` | view models | `CallViewModel` |
| `hostBridge.hangUp$`, `hangup()`, `userHangup$` | widget / host integration | `CallViewModel` | | `hostBridge.hangUp$`, `hangup()`, `userHangup$` | widget / host integration | `CallViewModel` |
| `ElementCallError` | carries Element Call translation keys | the view model wraps `MatrixRTCError`, reading `cause` | | `ElementCallError` | carries Element Call translation keys | the view model wraps `MatrixRTCError`, reading `cause` |
@@ -645,11 +656,8 @@ public API only.
## Open decisions ## Open decisions
- **`LocalMediaInputs` shape.** Everything in it is an initial value; the - **Publishing before `join()`.** The initial requests are published at the
setters on the client and the controls on the local tracks change it from join; creating the tracks earlier, for a lobby preview, is still missing.
then on, because they have to report what the device allowed. Creating the
tracks before `join()` for a preview is still missing: today they are created
at the join.
- **`applicationData` after the join.** The option is read once. The client - **`applicationData` after the join.** The option is read once. The client
still updates `m.call.intent` from the camera state, as Element Call did; that still updates `m.call.intent` from the camera state, as Element Call did; that
moves to the host once the data can be updated. moves to the host once the data can be updated.
+3
View File
@@ -60,6 +60,9 @@ Please see LICENSE in the repository root for full details.
<button type="button" id="camera" aria-pressed="true" hidden> <button type="button" id="camera" aria-pressed="true" hidden>
Camera Camera
</button> </button>
<button type="button" id="screenShare" aria-pressed="false" hidden>
Share screen
</button>
<button type="button" id="leave" hidden>Leave</button> <button type="button" id="leave" hidden>Leave</button>
</form> </form>
<p id="status" data-testid="status"></p> <p id="status" data-testid="status"></p>
+32 -4
View File
@@ -36,6 +36,7 @@ const messages = document.getElementById("messages")!;
const buttons = { const buttons = {
microphone: document.getElementById("microphone") as HTMLButtonElement, microphone: document.getElementById("microphone") as HTMLButtonElement,
camera: document.getElementById("camera") as HTMLButtonElement, camera: document.getElementById("camera") as HTMLButtonElement,
screenShare: document.getElementById("screenShare") as HTMLButtonElement,
leave: document.getElementById("leave") as HTMLButtonElement, leave: document.getElementById("leave") as HTMLButtonElement,
}; };
@@ -74,8 +75,7 @@ async function start(
client, client,
room, room,
{ {
microphoneEnabled: true, publish: [{ source: "microphone" }, { source: "camera" }],
cameraEnabled: true,
audioOutputDeviceId$: constant(undefined), audioOutputDeviceId$: constant(undefined),
}, },
{ {
@@ -94,10 +94,13 @@ async function start(
rtcClient.join(); rtcClient.join();
toggle(buttons.microphone, async (enabled) => toggle(buttons.microphone, async (enabled) =>
rtcClient.setMicrophoneEnabled(enabled), setPublished(rtcClient, "microphone", enabled),
); );
toggle(buttons.camera, async (enabled) => toggle(buttons.camera, async (enabled) =>
rtcClient.setCameraEnabled(enabled), setPublished(rtcClient, "camera", enabled),
);
toggle(buttons.screenShare, async (enabled) =>
setPublished(rtcClient, "screenShare", enabled),
); );
buttons.leave.hidden = false; buttons.leave.hidden = false;
buttons.leave.onclick = (): void => { buttons.leave.onclick = (): void => {
@@ -115,6 +118,31 @@ async function start(
} }
} }
/**
* Turns one of our sources on or off: a microphone or camera is muted and
* unmuted while it exists and published when it does not; a screen share is
* published and unpublished.
*/
async function setPublished(
rtcClient: MatrixRTCClient,
source: "microphone" | "camera" | "screenShare",
enabled: boolean,
): Promise<boolean> {
const member = rtcClient.localMember$.value;
if (member === null) return !enabled;
const track = member.media$.value?.tracks$.value.find(
(t) => t.source === source,
);
if (track && source === "screenShare") {
if (!enabled) await member.unpublish(track.id);
return enabled;
}
if (track) return track.setEnabled(enabled);
if (!enabled) return false;
await member.publish({ source });
return true;
}
/** A pressed button means the source is on; the SDK says what it could do. */ /** A pressed button means the source is on; the SDK says what it could do. */
function toggle( function toggle(
button: HTMLButtonElement, button: HTMLButtonElement,
+34 -30
View File
@@ -57,23 +57,16 @@ export interface MatrixRTCClientOptions {
timings?: Partial<SessionTimings>; timings?: Partial<SessionTimings>;
/** Limits on what is published; LiveKit's defaults otherwise. */ /** Limits on what is published; LiveKit's defaults otherwise. */
mediaQuality?: MediaQuality; mediaQuality?: MediaQuality;
/** How the local tracks are captured and encoded. */
capture?: CaptureSettings;
/** Use this transport instead of asking the homeserver. */ /** Use this transport instead of asking the homeserver. */
transportUrl?: string; transportUrl?: string;
/** Use this transport when the homeserver advertises none. */ /** Use this transport when the homeserver advertises none. */
fallbackTransportUrl?: string; fallbackTransportUrl?: string;
} }
/** Capture and encoding settings for the local tracks, applied at creation. */ export interface AudioCaptureSettings {
export interface CaptureSettings {
audio?: {
echoCancellation?: boolean; echoCancellation?: boolean;
noiseSuppression?: boolean; noiseSuppression?: boolean;
autoGainControl?: boolean; autoGainControl?: boolean;
};
camera?: VideoCaptureSettings;
screenShare?: VideoCaptureSettings;
} }
export interface VideoCaptureSettings { export interface VideoCaptureSettings {
@@ -84,17 +77,29 @@ export interface VideoCaptureSettings {
} }
/** /**
* What the local member publishes when it joins. Initial values: from then on * One thing to publish: a source, where to capture it from and how to encode
* `setMicrophoneEnabled` and `setCameraEnabled` on the client, and the * it. Where the device and the settings are left out, the browser's and
* controls on the local tracks, change it. * LiveKit's defaults apply.
*/ */
export interface LocalMediaInputs { export type PublishRequest =
microphoneEnabled: boolean; | { source: "microphone"; deviceId?: string; capture?: AudioCaptureSettings }
cameraEnabled: boolean; | {
audioInputDeviceId?: string; source: "camera";
videoInputDeviceId?: string; deviceId?: string;
/** Background blur and the like. */ /** Background blur and the like. */
videoProcessor?: TrackProcessor<Track.Kind.Video>; processor?: TrackProcessor<Track.Kind.Video>;
capture?: VideoCaptureSettings;
}
| {
source: "screenShare";
/** Whether to capture the screen's audio too. Default true. */
audio?: boolean;
capture?: VideoCaptureSettings;
};
export interface LocalMediaInputs {
/** Published at the join; `publish` on the local member adds to it from then on. */
publish: PublishRequest[];
/** Undefined where the host routes audio itself, or to leave the browser's choice. */ /** Undefined where the host routes audio itself, or to leave the browser's choice. */
audioOutputDeviceId$: Behavior<string | undefined>; audioOutputDeviceId$: Behavior<string | undefined>;
} }
@@ -129,14 +134,6 @@ export interface MatrixRTCClient {
/** Null while connected, the first failing link otherwise. */ /** Null while connected, the first failing link otherwise. */
disconnectReason$: Behavior<DisconnectReason | null>; 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,11 +236,18 @@ export interface RemoteRTCMember extends RTCMember {
export interface LocalRTCMember extends RTCMember { export interface LocalRTCMember extends RTCMember {
local: true; local: true;
media$: Behavior<LocalMemberMedia | null>; media$: Behavior<LocalMemberMedia | null>;
sharingScreen$: Behavior<boolean>; /**
/** Null when the platform cannot share a screen. */ * Publishes a source and resolves with its track once it is in `tracks$`.
toggleScreenSharing: (() => void) | null; * Before the transport is connected the request is remembered and applied
screenShareError$: Behavior<Error | null>; * once it is. Rejects where the device could not be used, including the
dismissScreenShareError(): void; * user closing the picker. One publication per source: publishing a source
* again unmutes it.
*/
publish(
request: PublishRequest,
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
/** Removes one of our tracks; a screen share takes its audio with it. */
unpublish(id: string): Promise<void>;
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
+2 -2
View File
@@ -52,7 +52,7 @@ import {
import { MatrixRTCError } from "../errors"; import { MatrixRTCError } from "../errors";
import { LazyBehavior } from "../utils/LazyBehavior"; import { LazyBehavior } from "../utils/LazyBehavior";
const sources: Record<Track.Source, MediaSource> = { export const mediaSources: Record<Track.Source, MediaSource> = {
[Track.Source.Microphone]: "microphone", [Track.Source.Microphone]: "microphone",
[Track.Source.Camera]: "camera", [Track.Source.Camera]: "camera",
[Track.Source.ScreenShare]: "screenShare", [Track.Source.ScreenShare]: "screenShare",
@@ -114,7 +114,7 @@ export function createLivekitMediaTrack(
publication.isMuted, publication.isMuted,
); );
const base: MediaTrack = { const base: MediaTrack = {
source: sources[publication.source], source: mediaSources[publication.source],
kind: publication.kind === Track.Kind.Audio ? "audio" : "video", kind: publication.kind === Track.Kind.Audio ? "audio" : "video",
id: publication.trackSid, id: publication.trackSid,
muted$, muted$,
+23 -26
View File
@@ -23,12 +23,14 @@ 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 { import { type LocalMediaInputs, type ResolvedTransport } from "../api";
type CaptureSettings,
type LocalMediaInputs,
type ResolvedTransport,
} from "../api";
import { type MediaQuality } from "../config"; import { type MediaQuality } from "../config";
import {
audioCaptureOptions,
type PublishRequestFor,
videoCaptureOptions,
videoPublishOptions,
} from "./publishOptions";
export interface ConnectionFactory { export interface ConnectionFactory {
createConnection( createConnection(
@@ -93,7 +95,6 @@ export class LivekitConnectionFactory implements ConnectionFactory {
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 mediaQuality: MediaQuality | undefined,
private readonly capture: CaptureSettings | undefined,
) {} ) {}
public createConnection( public createConnection(
@@ -112,12 +113,7 @@ export class LivekitConnectionFactory implements ConnectionFactory {
scope, scope,
livekitRoomFactory: () => livekitRoomFactory: () =>
new LivekitRoom( new LivekitRoom(
roomOptions( roomOptions(this.localMedia, this.keyProvider, this.mediaQuality),
this.localMedia,
this.keyProvider,
this.mediaQuality,
this.capture,
),
), ),
ownMembershipIdentity, ownMembershipIdentity,
}, },
@@ -126,36 +122,28 @@ export class LivekitConnectionFactory implements ConnectionFactory {
} }
} }
/** The initial requests become the room's defaults, so that one permission prompt covers both. */
function roomOptions( function roomOptions(
localMedia: LocalMediaInputs, localMedia: LocalMediaInputs,
keyProvider: BaseKeyProvider | undefined, keyProvider: BaseKeyProvider | undefined,
mediaQuality: MediaQuality | undefined, mediaQuality: MediaQuality | undefined,
capture: CaptureSettings | undefined,
): RoomOptions { ): RoomOptions {
const base = buildLiveKitOptions(mediaQuality); const base = buildLiveKitOptions(mediaQuality);
const camera = capture?.camera; const microphone = initialRequest(localMedia, "microphone");
const camera = initialRequest(localMedia, "camera");
return { return {
...base, ...base,
videoCaptureDefaults: { videoCaptureDefaults: {
...base.videoCaptureDefaults, ...base.videoCaptureDefaults,
deviceId: localMedia.videoInputDeviceId, ...(camera && videoCaptureOptions(camera)),
processor: localMedia.videoProcessor,
...(camera?.resolution && { resolution: camera.resolution }),
}, },
publishDefaults: { publishDefaults: {
...base.publishDefaults, ...base.publishDefaults,
...(camera?.maxBitrate !== undefined && { ...(camera && videoPublishOptions(camera.capture)),
videoEncoding: {
maxBitrate: camera.maxBitrate,
maxFramerate: camera.maxFramerate,
},
}),
...(camera?.codec && { videoCodec: camera.codec }),
}, },
audioCaptureDefaults: { audioCaptureDefaults: {
...base.audioCaptureDefaults, ...base.audioCaptureDefaults,
deviceId: localMedia.audioInputDeviceId, ...(microphone && audioCaptureOptions(microphone)),
...capture?.audio,
}, },
audioOutput: { deviceId: localMedia.audioOutputDeviceId$.value }, 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
@@ -163,3 +151,12 @@ function roomOptions(
e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() }, e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() },
}; };
} }
function initialRequest<S extends "microphone" | "camera">(
localMedia: LocalMediaInputs,
source: S,
): PublishRequestFor<S> | undefined {
return localMedia.publish.find(
(request): request is PublishRequestFor<S> => request.source === source,
);
}
+56 -121
View File
@@ -8,11 +8,8 @@ Please see LICENSE in the repository root for full details.
import { observeParticipantEvents } from "@livekit/components-core"; import { observeParticipantEvents } from "@livekit/components-core";
import { import {
type LocalParticipant, type LocalParticipant,
type Participant, type LocalTrackPublication,
ParticipantEvent, ParticipantEvent,
type ScreenShareCaptureOptions,
Track,
type TrackPublishOptions,
} from "livekit-client"; } from "livekit-client";
import { type MatrixClient } from "matrix-js-sdk"; 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";
@@ -29,6 +26,8 @@ import {
combineLatest, combineLatest,
concat, concat,
distinctUntilChanged, distinctUntilChanged,
filter,
firstValueFrom,
map, map,
NEVER, NEVER,
type Observable, type Observable,
@@ -49,7 +48,7 @@ import {
import { import {
type DisconnectReason, type DisconnectReason,
type MediaSource, type MediaSource,
type VideoCaptureSettings, type PublishRequest,
} from "../api"; } from "../api";
import { import {
FailToStartLivekitConnection, FailToStartLivekitConnection,
@@ -105,7 +104,6 @@ interface Props {
matrixRTCMode: MatrixRTCMode; matrixRTCMode: MatrixRTCMode;
timings: SessionTimings; timings: SessionTimings;
desired: DesiredMedia; desired: DesiredMedia;
screenShare?: VideoCaptureSettings;
logger: Logger; logger: Logger;
} }
@@ -121,14 +119,11 @@ export interface LocalMembership {
/** Connected once, and currently not. */ /** Connected once, and currently not. */
reconnecting$: Behavior<boolean>; reconnecting$: Behavior<boolean>;
disconnectReason$: Behavior<DisconnectReason | null>; disconnectReason$: Behavior<DisconnectReason | null>;
setMicrophoneEnabled: (enabled: boolean) => Promise<boolean>; /** Resolves with the publication, once a publisher has made it. */
setCameraEnabled: (enabled: boolean) => Promise<boolean>; publish: (request: PublishRequest) => Promise<LocalTrackPublication>;
unpublish: (source: MediaSource) => Promise<void>;
/** Mutes or unmutes one of our publications; the result is the state that resulted. */ /** Mutes or unmutes one of our publications; the result is the state that resulted. */
setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>; setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>;
sharingScreen$: Behavior<boolean>;
toggleScreenSharing: (() => void) | null;
screenShareError$: Behavior<Error | null>;
dismissScreenShareError: () => void;
} }
/** /**
@@ -152,7 +147,6 @@ export function createLocalMembership$({
matrixRTCMode, matrixRTCMode,
timings, timings,
desired, desired,
screenShare,
logger: parentLogger, logger: parentLogger,
}: Props): LocalMembership { }: Props): LocalMembership {
const logger = parentLogger.getChild("[LocalMember]"); const logger = parentLogger.getChild("[LocalMember]");
@@ -222,7 +216,7 @@ export function createLocalMembership$({
if (publisher === null) return; if (publisher === null) return;
try { try {
if (shouldPublish) { if (shouldPublish) {
publisher.createAndSetupTracks(); publisher.start();
await publisher.startPublishing(); await publisher.startPublishing();
} else if (publisher.shouldPublish) await publisher.stopPublishing(); } else if (publisher.shouldPublish) await publisher.stopPublishing();
} catch (e) { } catch (e) {
@@ -398,74 +392,62 @@ export function createLocalMembership$({
} }
}); });
const publish = async (
request: PublishRequest,
): Promise<LocalTrackPublication> => {
desired.set(request.source, { request, enabled: true });
const publisher = publisher$.value;
try {
// Before the start, the publisher publishes everything desired itself
if (publisher?.started) {
const publication = await publisher.publish(request);
if (publication) return publication;
}
return await firstValueFrom(publication$(request.source));
} catch (e) {
desired.delete(request.source);
throw toMatrixRTCError(e);
}
};
const publication$ = (
source: MediaSource,
): Observable<LocalTrackPublication> =>
participant$.pipe(
switchMap((participant) =>
participant === null
? NEVER
: observeParticipantEvents(
participant,
ParticipantEvent.LocalTrackPublished,
).pipe(
map(() =>
participant.getTrackPublication(livekitSources[source]),
),
),
),
filter((publication) => publication !== undefined),
);
const unpublish = async (source: MediaSource): Promise<void> => {
desired.delete(source);
await publisher$.value?.unpublish(source);
};
const setEnabled = async ( const setEnabled = async (
source: MediaSource, source: MediaSource,
enabled: boolean, enabled: boolean,
): Promise<boolean> => { ): Promise<boolean> => {
if (source !== "microphone" && source !== "camera") const wanted = desired.get(source);
return setPublicationEnabled(source, enabled); if (wanted) wanted.enabled = enabled;
const desired$ =
source === "microphone" ? desired.microphone$ : desired.camera$;
desired$.next(enabled);
// Without a publisher the request waits for the tracks to be created // Without a publisher the request waits for the tracks to be created
const publisher = publisher$.value; const publisher = publisher$.value;
if (publisher === null) return enabled; if (publisher === null) return enabled;
const result = await publisher.setEnabled( const result = await publisher.setEnabled(source, enabled);
source === "microphone" ? Track.Source.Microphone : Track.Source.Camera, if (wanted && result !== enabled) wanted.enabled = result;
enabled,
);
if (result !== enabled) desired$.next(result);
return result; return result;
}; };
// A publication the publisher does not recreate on reconnection, so there
// is nothing to remember for it
const setPublicationEnabled = async (
source: MediaSource,
enabled: boolean,
): Promise<boolean> => {
const publication = participant$.value?.getTrackPublication(
livekitSources[source],
);
if (!publication) return false;
if (enabled) await publication.unmute();
else await publication.mute();
return !publication.isMuted;
};
const sharingScreen$ = scope.behavior(
participant$.pipe(
switchMap((p) => (p === null ? of(false) : observeSharingScreen$(p))),
),
);
const screenShareError$ = new BehaviorSubject<Error | null>(null);
const toggleScreenSharing =
"getDisplayMedia" in (navigator.mediaDevices ?? {})
? (): void => {
const participant = participant$.value;
if (participant === null) return;
const enable = !sharingScreen$.value;
participant
.setScreenShareEnabled(
enable,
screenShareCaptureOptions(screenShare),
screenSharePublishOptions(screenShare),
)
.catch((e: unknown) => {
logger.error(
`Screen share ${enable ? "start" : "stop"} failed`,
e,
);
// The user closing the picker is not an error worth showing
if (e instanceof DOMException && e.name === "NotAllowedError")
return;
screenShareError$.next(
e instanceof Error ? e : new Error(String(e)),
);
});
}
: null;
return { return {
requestJoinAndPublish: () => joinRequested$.next(true), requestJoinAndPublish: () => joinRequested$.next(true),
requestDisconnect: () => joinRequested$.next(false), requestDisconnect: () => joinRequested$.next(false),
@@ -476,13 +458,9 @@ export function createLocalMembership$({
connected$, connected$,
reconnecting$, reconnecting$,
disconnectReason$, disconnectReason$,
setMicrophoneEnabled: async (enabled) => setEnabled("microphone", enabled), publish,
setCameraEnabled: async (enabled) => setEnabled("camera", enabled), unpublish,
setEnabled, setEnabled,
sharingScreen$,
toggleScreenSharing,
screenShareError$,
dismissScreenShareError: () => screenShareError$.next(null),
}; };
} }
@@ -511,46 +489,3 @@ async function checkDelegationSupport(
return false; return false;
} }
} }
function screenShareCaptureOptions(
settings: VideoCaptureSettings | undefined,
): ScreenShareCaptureOptions {
return {
// No echo cancellation: it would cancel the other members' voices out of
// the shared audio
audio: {
autoGainControl: false,
noiseSuppression: false,
voiceIsolation: false,
},
selfBrowserSurface: "include",
surfaceSwitching: "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> {
return observeParticipantEvents(
participant,
ParticipantEvent.TrackPublished,
ParticipantEvent.TrackUnpublished,
ParticipantEvent.LocalTrackPublished,
ParticipantEvent.LocalTrackUnpublished,
).pipe(map((p) => p.isScreenShareEnabled));
}
+7 -16
View File
@@ -15,14 +15,7 @@ import {
import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager"; 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 { import { combineLatest, from, fromEvent, map, type Observable } from "rxjs";
BehaviorSubject,
combineLatest,
from,
fromEvent,
map,
type Observable,
} from "rxjs";
import { defaultSessionTimings, MatrixRTCMode } from "../config"; import { defaultSessionTimings, MatrixRTCMode } from "../config";
import { type ObservableScope } from "../reactive/ObservableScope"; import { type ObservableScope } from "../reactive/ObservableScope";
@@ -123,10 +116,12 @@ export function createMatrixRTCClient(
}), }),
); );
const desired: DesiredMedia = { const desired: DesiredMedia = new Map(
microphone$: new BehaviorSubject(localMedia.microphoneEnabled), localMedia.publish.map((request) => [
camera$: new BehaviorSubject(localMedia.cameraEnabled), request.source,
}; { request, enabled: true },
]),
);
const connectionManager = createConnectionManager$({ const connectionManager = createConnectionManager$({
scope, scope,
@@ -136,7 +131,6 @@ export function createMatrixRTCClient(
localMedia, localMedia,
keyProvider, keyProvider,
options.mediaQuality, options.mediaQuality,
options.capture,
), ),
localTransport$, localTransport$,
remoteTransports$: transports$, remoteTransports$: transports$,
@@ -194,7 +188,6 @@ export function createMatrixRTCClient(
matrixRTCMode, matrixRTCMode,
timings, timings,
desired, desired,
screenShare: options.capture?.screenShare,
logger, logger,
}); });
@@ -276,8 +269,6 @@ export function createMatrixRTCClient(
connected$: localMembership.connected$, connected$: localMembership.connected$,
reconnecting$: localMembership.reconnecting$, reconnecting$: localMembership.reconnecting$,
disconnectReason$: localMembership.disconnectReason$, 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$,
+31 -9
View File
@@ -11,7 +11,15 @@ import {
type Room as LivekitRoom, type Room as LivekitRoom,
} from "livekit-client"; } from "livekit-client";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc"; import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, distinctUntilChanged, map } from "rxjs"; import {
combineLatest,
distinctUntilChanged,
filter,
firstValueFrom,
map,
of,
switchMap,
} from "rxjs";
import { type Behavior } from "../reactive/Behavior"; import { type Behavior } from "../reactive/Behavior";
import { type ObservableScope } from "../reactive/ObservableScope"; import { type ObservableScope } from "../reactive/ObservableScope";
@@ -80,10 +88,7 @@ export function createLocalRTCMember(
localMembership: LocalMembership, localMembership: LocalMembership,
context: MemberContext, context: MemberContext,
): LocalRTCMember { ): LocalRTCMember {
return { const media$ = mediaFor(
...createRTCMember(scope, membership$, context),
local: true,
media$: mediaFor(
scope, scope,
localMembership.participant$, localMembership.participant$,
localMembership.connection$, localMembership.connection$,
@@ -95,11 +100,28 @@ export function createLocalRTCMember(
context.encryptionSystem, context.encryptionSystem,
localMembership.setEnabled, localMembership.setEnabled,
), ),
);
return {
...createRTCMember(scope, membership$, context),
local: true,
media$,
publish: async (request) => {
const { trackSid } = await localMembership.publish(request);
return firstValueFrom(
media$.pipe(
switchMap((media) => media?.tracks$ ?? of([])),
map((tracks) => tracks.find((track) => track.id === trackSid)),
filter((track) => track !== undefined),
), ),
sharingScreen$: localMembership.sharingScreen$, );
toggleScreenSharing: localMembership.toggleScreenSharing, },
screenShareError$: localMembership.screenShareError$, unpublish: async (id) => {
dismissScreenShareError: localMembership.dismissScreenShareError, const track = media$.value?.tracks$.value.find((t) => t.id === id);
if (track === undefined) return;
await localMembership.unpublish(
track.source === "screenShareAudio" ? "screenShare" : track.source,
);
},
}; };
} }
+110 -73
View File
@@ -10,20 +10,33 @@ import {
type LocalTrackPublication, type LocalTrackPublication,
ParticipantEvent, ParticipantEvent,
type Room as LivekitRoom, type Room as LivekitRoom,
Track, type 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 { type BehaviorSubject } from "rxjs";
import { ObservableScope } from "../reactive/ObservableScope"; import { ObservableScope } from "../reactive/ObservableScope";
import { type LocalMediaInputs } from "../api"; import {
type LocalMediaInputs,
type MediaSource,
type PublishRequest,
} from "../api";
import { livekitSources, mediaSources } from "../media/LivekitMediaTrack";
import {
audioCaptureOptions,
screenShareCaptureOptions,
screenSharePublishOptions,
videoCaptureOptions,
videoPublishOptions,
} from "./publishOptions";
/** What the host last asked for, kept across publishers. */ /** What the host asked to publish, kept across publishers for a reconnection. */
export interface DesiredMedia { export interface DesiredPublication {
microphone$: BehaviorSubject<boolean>; request: PublishRequest;
camera$: BehaviorSubject<boolean>; enabled: boolean;
} }
export type DesiredMedia = Map<MediaSource, DesiredPublication>;
/** /**
* Publishes the local media on one LiveKit room, following the host's * Publishes the local media on one LiveKit room, following the host's
* requests. * requests.
@@ -35,7 +48,8 @@ export interface DesiredMedia {
*/ */
export class Publisher { export class Publisher {
public shouldPublish = false; public shouldPublish = false;
private tracksRequested = false; /** Whether the initial requests have been published; `publish` is immediate from then on. */
public started = false;
private readonly scope = new ObservableScope(); private readonly scope = new ObservableScope();
private readonly room: LivekitRoom; private readonly room: LivekitRoom;
@@ -64,53 +78,106 @@ export class Publisher {
this.onLocalTrackPublished, this.onLocalTrackPublished,
); );
try { try {
await this.stopTracks(); for (const { track } of this.publications())
if (track) await this.room.localParticipant.unpublishTrack(track, true);
} catch (e) { } catch (e) {
this.logger.error("Failed to stop the local tracks", e); this.logger.error("Failed to stop the local tracks", e);
} }
} }
/** /** Publishes what the host asked for, less what it has muted since. Safe to call more than once. */
* Creates the microphone and camera tracks the host asked for. Both are public start(): void {
* enabled in one call so that the browser asks for permission once. Safe to if (this.started) return;
* call more than once. this.started = true;
*/ const requests = [...this.desired.values()]
public createAndSetupTracks(): void { .filter(({ enabled }) => enabled)
if (this.tracksRequested) return; .map(({ request }) => request);
this.tracksRequested = true; const sources = new Set(requests.map((request) => request.source));
// Microphone and camera in one call so that the browser asks for
// permission once; the room defaults carry their options. LiveKit
// resolves these once the track is published, which may block on the
// connection; LocalTrackPublished is what tells us a track exists.
if (sources.has("microphone") && sources.has("camera"))
void this.room.localParticipant.enableCameraAndMicrophone();
for (const request of requests)
if (
request.source === "screenShare" ||
!(sources.has("microphone") && sources.has("camera"))
)
this.publish(request).catch((e) => {
this.logger.error(`Failed to publish the ${request.source}`, e);
});
}
/** Publishes one request and resolves with its publication once LiveKit has it. */
public async publish(
request: PublishRequest,
): Promise<LocalTrackPublication | undefined> {
const participant = this.room.localParticipant; const participant = this.room.localParticipant;
const audio = this.desired.microphone$.value; let publication: LocalTrackPublication | undefined;
const video = this.desired.camera$.value; if (request.source === "microphone")
// LiveKit resolves these once the track is published, which may block on publication = await participant.setMicrophoneEnabled(
// the connection; LocalTrackPublished is what tells us a track exists. true,
if (audio && video) void participant.enableCameraAndMicrophone(); audioCaptureOptions(request),
else if (audio) void participant.setMicrophoneEnabled(true); );
else if (video) void participant.setCameraEnabled(true); else if (request.source === "camera")
publication = await participant.setCameraEnabled(
true,
videoCaptureOptions(request),
videoPublishOptions(request.capture),
);
else
publication = await participant.setScreenShareEnabled(
true,
screenShareCaptureOptions(request),
screenSharePublishOptions(request.capture),
);
// Until the member has joined, the upstream has to stay paused
if (!this.shouldPublish) await this.pauseUpstreams();
return publication;
}
public async unpublish(source: MediaSource): Promise<void> {
const participant = this.room.localParticipant;
// Takes the screen share audio with it
if (source === "screenShare") {
await participant.setScreenShareEnabled(false);
return;
}
const track = participant.getTrackPublication(
livekitSources[source],
)?.track;
if (track) await participant.unpublishTrack(track, true);
} }
/** /**
* Enables or disables a source and reports what resulted: the request where * Mutes or unmutes a source and reports what resulted: the request where
* it succeeded, the state LiveKit is left in where it did not (a denied * it succeeded, the state LiveKit is left in where it did not (a denied
* permission, a missing device). * permission, a missing device).
*/ */
public async setEnabled( public async setEnabled(
source: Track.Source.Microphone | Track.Source.Camera, source: MediaSource,
enabled: boolean, enabled: boolean,
): Promise<boolean> { ): Promise<boolean> {
const participant = this.room.localParticipant; const participant = this.room.localParticipant;
const livekitSource = livekitSources[source];
try { try {
if (source === Track.Source.Microphone) if (source === "microphone")
await participant.setMicrophoneEnabled(enabled); await participant.setMicrophoneEnabled(enabled);
else await participant.setCameraEnabled(enabled); else if (source === "camera") await participant.setCameraEnabled(enabled);
else {
const publication = participant.getTrackPublication(livekitSource);
if (!publication) return false;
if (enabled) await publication.unmute();
else await publication.mute();
}
// Unmuting restarts the upstream; until the member has joined, it has // Unmuting restarts the upstream; until the member has joined, it has
// to stay paused // to stay paused
if (enabled && !this.shouldPublish) await this.pauseUpstreams([source]); if (enabled && !this.shouldPublish) await this.pauseUpstreams();
return enabled; return enabled;
} catch (e) { } catch (e) {
this.logger.error(`Failed to set ${source} enabled=${enabled}`, e); this.logger.error(`Failed to set ${source} enabled=${enabled}`, e);
return source === Track.Source.Microphone return !(participant.getTrackPublication(livekitSource)?.isMuted ?? true);
? participant.isMicrophoneEnabled
: participant.isCameraEnabled;
} }
} }
@@ -119,64 +186,34 @@ export class Publisher {
this.shouldPublish = true; this.shouldPublish = true;
// Enabling a track does not resume an upstream that was paused while it // Enabling a track does not resume an upstream that was paused while it
// was already enabled, so it is done explicitly // was already enabled, so it is done explicitly
await this.resumeUpstreams([Track.Source.Microphone, Track.Source.Camera]); for (const { track } of this.publications())
if (track?.isUpstreamPaused) await track.resumeUpstream();
} }
public async stopPublishing(): Promise<void> { public async stopPublishing(): Promise<void> {
this.shouldPublish = false; this.shouldPublish = false;
await this.pauseUpstreams([ await this.pauseUpstreams();
Track.Source.Microphone,
Track.Source.Camera,
Track.Source.ScreenShare,
]);
}
private async stopTracks(): Promise<void> {
const participant = this.room.localParticipant;
for (const source of [
Track.Source.Microphone,
Track.Source.Camera,
Track.Source.ScreenShare,
]) {
const track = participant.getTrackPublication(source)?.track;
if (track) await participant.unpublishTrack(track, true);
}
} }
private onLocalTrackPublished(publication: LocalTrackPublication): void { private onLocalTrackPublished(publication: LocalTrackPublication): void {
this.logger.info(`Local ${publication.source} track published`); this.logger.info(`Local ${publication.source} track published`);
if (!this.shouldPublish) if (!this.shouldPublish)
this.pauseUpstreams([publication.source]).catch((e) => { this.pauseUpstreams().catch((e) => {
this.logger.error("Failed to pause the upstream", e); this.logger.error("Failed to pause the upstream", e);
}); });
// The host may have changed its mind while the track was being created // The host may have changed its mind while the track was being created
const desired = const source = mediaSources[publication.source as Track.Source];
publication.source === Track.Source.Microphone if (this.desired.get(source)?.enabled === false)
? this.desired.microphone$.value void this.setEnabled(source, false);
: publication.source === Track.Source.Camera
? this.desired.camera$.value
: undefined;
if (desired === false)
void this.setEnabled(
publication.source as Track.Source.Microphone | Track.Source.Camera,
false,
);
} }
private async pauseUpstreams(sources: Track.Source[]): Promise<void> { private async pauseUpstreams(): Promise<void> {
for (const source of sources) { for (const { track } of this.publications())
const track =
this.room.localParticipant.getTrackPublication(source)?.track;
if (track && !track.isUpstreamPaused) await track.pauseUpstream(); if (track && !track.isUpstreamPaused) await track.pauseUpstream();
} }
}
private async resumeUpstreams(sources: Track.Source[]): Promise<void> { private publications(): LocalTrackPublication[] {
for (const source of sources) { return [...this.room.localParticipant.trackPublications.values()];
const track =
this.room.localParticipant.getTrackPublication(source)?.track;
if (track?.isUpstreamPaused) await track.resumeUpstream();
}
} }
private followAudioOutput(): void { private followAudioOutput(): void {
+91
View File
@@ -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 {
type AudioCaptureOptions,
type ScreenShareCaptureOptions,
type TrackPublishOptions,
type VideoCaptureOptions,
} from "livekit-client";
import { type PublishRequest, type VideoCaptureSettings } from "../api";
export type PublishRequestFor<S extends PublishRequest["source"]> = Extract<
PublishRequest,
{ source: S }
>;
/** LiveKit fills in the room defaults for whatever a request leaves out. */
export function audioCaptureOptions({
deviceId,
capture,
}: PublishRequestFor<"microphone">): AudioCaptureOptions | undefined {
if (deviceId === undefined && capture === undefined) return undefined;
return { ...(deviceId && { deviceId }), ...capture };
}
export function videoCaptureOptions({
deviceId,
processor,
capture,
}: PublishRequestFor<"camera">): VideoCaptureOptions | undefined {
if (!deviceId && !processor && !capture?.resolution) return undefined;
return {
...(deviceId && { deviceId }),
...(processor && { processor }),
...(capture?.resolution && { resolution: capture.resolution }),
};
}
export function videoPublishOptions(
settings: VideoCaptureSettings | undefined,
): TrackPublishOptions | undefined {
if (settings === undefined) return undefined;
return {
...(settings.maxBitrate !== undefined && {
videoEncoding: {
maxBitrate: settings.maxBitrate,
maxFramerate: settings.maxFramerate,
},
}),
...(settings.codec && { videoCodec: settings.codec }),
};
}
export function screenShareCaptureOptions({
audio = true,
capture,
}: PublishRequestFor<"screenShare">): ScreenShareCaptureOptions {
return {
// No echo cancellation: it would cancel the other members' voices out of
// the shared audio
audio: audio && {
autoGainControl: false,
noiseSuppression: false,
voiceIsolation: false,
},
selfBrowserSurface: "include",
surfaceSwitching: "include",
systemAudio: "include",
...(capture?.resolution && { resolution: capture.resolution }),
};
}
export 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 }),
};
}