From 5a43eaf5a62688bbb9def8d914efc39ca62d5124 Mon Sep 17 00:00:00 2001 From: "Timo K." Date: Sun, 4 Oct 2026 12:40:43 +0200 Subject: [PATCH] Give the local tracks their controls Our own tracks are LocalAudioMediaTrack and LocalVideoMediaTrack, with setEnabled and setDevice, and on video facingMode$, switchFacingMode and setProcessor. switchCamera on the local media goes, as do the device and processor behaviors in LocalMediaInputs: a host calls the track when its setting changes instead of the SDK watching behaviors it does not own. The input device ids and the processor stay as initial values for the tracks created at the join. setMicrophoneEnabled and setCameraEnabled stay for the case of no track yet. Co-Authored-By: Claude Fable 5.1 --- sdk/README.md | 3 - sdk/SdkArchitecture.md | 57 +++++++++----- sdk/dev/main.ts | 3 - sdk/src/api.ts | 53 +++++++++---- sdk/src/media/LivekitMediaTrack.test.ts | 27 ++++++- sdk/src/media/LivekitMediaTrack.ts | 100 +++++++++++++++++++++--- sdk/src/media/LivekitMemberMedia.ts | 81 ++++++++++--------- sdk/src/session/ConnectionFactory.ts | 6 +- sdk/src/session/LocalMember.ts | 44 ++++++++--- sdk/src/session/Members.ts | 1 + sdk/src/session/Publisher.ts | 62 +++------------ 11 files changed, 279 insertions(+), 158 deletions(-) diff --git a/sdk/README.md b/sdk/README.md index c3e25e4d8..65f77b3aa 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -32,10 +32,7 @@ const rtcClient = createMatrixRTCClient( { microphoneEnabled: true, cameraEnabled: true, - audioInputDeviceId$: constant(undefined), - videoInputDeviceId$: constant(undefined), audioOutputDeviceId$: constant(undefined), - videoProcessor$: constant(undefined), }, { encryptionSystem: { kind: E2eeType.PER_PARTICIPANT }, diff --git a/sdk/SdkArchitecture.md b/sdk/SdkArchitecture.md index 683a0c4fd..5fffba2eb 100644 --- a/sdk/SdkArchitecture.md +++ b/sdk/SdkArchitecture.md @@ -198,20 +198,20 @@ export interface MatrixRTCClientOptions { } /** - * 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. Device enumeration, - * permission prompts and the lobby preview stay with the host. + * What the local member publishes when it joins. Initial values: from then on + * `setMicrophoneEnabled` and `setCameraEnabled` on the client, and the + * controls on the local tracks, change it. Device enumeration, permission + * prompts and the lobby preview stay with the host. */ export interface LocalMediaInputs { microphoneEnabled: boolean; cameraEnabled: boolean; - audioInputDeviceId$: Behavior; - videoInputDeviceId$: Behavior; + audioInputDeviceId?: string; + videoInputDeviceId?: string; + /** Background blur and the like. */ + videoProcessor?: TrackProcessor; /** Undefined where the host routes audio itself, or to leave the browser's choice. */ audioOutputDeviceId$: Behavior; - /** Background blur and the like. */ - videoProcessor$: Behavior | undefined>; } /** @@ -500,8 +500,28 @@ export interface AudioMediaTrack extends MediaTrack { export interface VideoMediaTrack extends MediaTrack { kind: "video"; - /** For mirroring and switching cameras. Undefined for remote tracks. */ - facingMode$?: Behavior<"user" | "environment" | undefined>; +} + +/** The controls a member has over a track it publishes itself. */ +export interface LocalMediaTrack { + /** Mutes or unmutes. Resolves with the state that resulted. */ + setEnabled(enabled: boolean): Promise; + /** Captures from another device. Rejects for a screen share, which has none. */ + setDevice(deviceId: string): Promise; +} + +export interface LocalAudioMediaTrack + extends AudioMediaTrack, LocalMediaTrack {} + +export interface LocalVideoMediaTrack extends VideoMediaTrack, LocalMediaTrack { + /** For mirroring; undefined where the camera does not say which way it faces. */ + facingMode$: Behavior<"user" | "environment" | undefined>; + /** Restarts the camera facing the other way; resolves with the device now in use. */ + switchFacingMode(): Promise; + /** Background blur and the like; undefined removes the processor. */ + setProcessor( + processor: TrackProcessor | undefined, + ): Promise; } export type EncryptionError = "MissingKey" | "InvalidKey"; @@ -532,12 +552,7 @@ export function trackBySource$( export interface LocalMemberMedia extends MemberMedia { 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; + tracks$: Behavior<(LocalAudioMediaTrack | LocalVideoMediaTrack)[]>; } ``` @@ -630,11 +645,11 @@ public API only. ## Open decisions -- **`LocalMediaInputs` shape.** The enabled flags are initial values and the - setters on the client change them, because the setters have to report what - the device allowed; the device ids and the processor stay behaviors. Creating - the tracks before `join()` for a preview is still missing: today they are - created at the join. +- **`LocalMediaInputs` shape.** Everything in it is an initial value; the + setters on the client and the controls on the local tracks change it from + 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 still updates `m.call.intent` from the camera state, as Element Call did; that moves to the host once the data can be updated. diff --git a/sdk/dev/main.ts b/sdk/dev/main.ts index 75445c2db..f73cc206a 100644 --- a/sdk/dev/main.ts +++ b/sdk/dev/main.ts @@ -76,10 +76,7 @@ async function start( { microphoneEnabled: true, cameraEnabled: true, - audioInputDeviceId$: constant(undefined), - videoInputDeviceId$: constant(undefined), audioOutputDeviceId$: constant(undefined), - videoProcessor$: constant(undefined), }, { encryptionSystem: { kind: E2eeType.PER_PARTICIPANT }, diff --git a/sdk/src/api.ts b/sdk/src/api.ts index 4b9f0a1d3..aa468f583 100644 --- a/sdk/src/api.ts +++ b/sdk/src/api.ts @@ -84,19 +84,19 @@ export interface VideoCaptureSettings { } /** - * 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. + * What the local member publishes when it joins. Initial values: from then on + * `setMicrophoneEnabled` and `setCameraEnabled` on the client, and the + * controls on the local tracks, change it. */ export interface LocalMediaInputs { microphoneEnabled: boolean; cameraEnabled: boolean; - audioInputDeviceId$: Behavior; - videoInputDeviceId$: Behavior; + audioInputDeviceId?: string; + videoInputDeviceId?: string; + /** Background blur and the like. */ + videoProcessor?: TrackProcessor; /** Undefined where the host routes audio itself, or to leave the browser's choice. */ audioOutputDeviceId$: Behavior; - /** Background blur and the like. */ - videoProcessor$: Behavior | undefined>; } export type ConnectionStatus = @@ -300,8 +300,35 @@ export interface AudioMediaTrack extends MediaTrack { export interface VideoMediaTrack extends MediaTrack { kind: "video"; - /** Undefined for remote tracks. */ - facingMode$?: Behavior<"user" | "environment" | undefined>; +} + +/** The controls a member has over a track it publishes itself. */ +export interface LocalMediaTrack { + /** + * Mutes or unmutes. Resolves with the state that resulted, which differs + * from the request where the device could not be used. + */ + setEnabled(enabled: boolean): Promise; + /** Captures from another device. Rejects for a screen share, which has none. */ + setDevice(deviceId: string): Promise; +} + +export interface LocalAudioMediaTrack + extends AudioMediaTrack, LocalMediaTrack {} + +export interface LocalVideoMediaTrack extends VideoMediaTrack, LocalMediaTrack { + /** For mirroring; undefined where the camera does not say which way it faces. */ + facingMode$: Behavior<"user" | "environment" | undefined>; + /** + * 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 where the facing mode is unknown. + */ + switchFacingMode(): Promise; + /** Background blur and the like; undefined removes the processor. */ + setProcessor( + processor: TrackProcessor | undefined, + ): Promise; } export type EncryptionError = "MissingKey" | "InvalidKey"; @@ -325,11 +352,5 @@ export interface MemberMedia { export interface LocalMemberMedia extends MemberMedia { 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; + tracks$: Behavior<(LocalAudioMediaTrack | LocalVideoMediaTrack)[]>; } diff --git a/sdk/src/media/LivekitMediaTrack.test.ts b/sdk/src/media/LivekitMediaTrack.test.ts index 87f97e6e2..b696dafe4 100644 --- a/sdk/src/media/LivekitMediaTrack.test.ts +++ b/sdk/src/media/LivekitMediaTrack.test.ts @@ -7,6 +7,7 @@ Please see LICENSE in the repository root for full details. import { EventEmitter } from "events"; import { + type LocalParticipant, ParticipantEvent, RemoteAudioTrack, type RemoteParticipant, @@ -19,7 +20,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ObservableScope } from "../reactive/ObservableScope"; import { type AudioMediaTrack } from "../api"; -import { createLivekitMediaTrack } from "./LivekitMediaTrack"; +import { + createLivekitMediaTrack, + createLocalLivekitMediaTrack, +} from "./LivekitMediaTrack"; describe("createLivekitMediaTrack", () => { let scope: ObservableScope; @@ -127,6 +131,27 @@ describe("createLivekitMediaTrack", () => { emitter.emit(ParticipantEvent.TrackMuted, publication); expect(track.isActive$.value).toBe(false); }); + + it("gives our own track its controls", async () => { + const { participant, publication, room } = fakes({ + kind: Track.Kind.Audio, + }); + const setEnabled = vi.fn().mockResolvedValue(false); + const switchActiveDevice = vi.fn().mockResolvedValue(true); + const track = createLocalLivekitMediaTrack( + scope, + Object.assign(participant, { + isLocal: true, + }) as unknown as LocalParticipant, + publication as unknown as TrackPublication, + Object.assign(room, { switchActiveDevice }), + setEnabled, + ); + await expect(track.setEnabled(false)).resolves.toBe(false); + expect(setEnabled).toHaveBeenCalledWith("microphone", false); + await track.setDevice("mic-2"); + expect(switchActiveDevice).toHaveBeenCalledWith("audioinput", "mic-2"); + }); }); type FakeTrack = Track & { diff --git a/sdk/src/media/LivekitMediaTrack.ts b/sdk/src/media/LivekitMediaTrack.ts index bd60504f7..b71be7ed2 100644 --- a/sdk/src/media/LivekitMediaTrack.ts +++ b/sdk/src/media/LivekitMediaTrack.ts @@ -12,6 +12,7 @@ import { } from "@livekit/components-core"; import { facingModeFromLocalTrack, + type LocalParticipant, LocalTrack, LocalVideoTrack, type Participant, @@ -41,11 +42,14 @@ import { type Behavior } from "../reactive/Behavior"; import { type ObservableScope } from "../reactive/ObservableScope"; import { type AudioMediaTrack, + type LocalAudioMediaTrack, + type LocalVideoMediaTrack, type MediaSource, type MediaStreamStats, type MediaTrack, type VideoMediaTrack, } from "../api"; +import { MatrixRTCError } from "../errors"; import { LazyBehavior } from "../utils/LazyBehavior"; const sources: Record = { @@ -56,6 +60,14 @@ const sources: Record = { [Track.Source.Unknown]: "unknown", }; +export const livekitSources: Record = { + microphone: Track.Source.Microphone, + camera: Track.Source.Camera, + screenShare: Track.Source.ScreenShare, + screenShareAudio: Track.Source.ScreenShareAudio, + unknown: Track.Source.Unknown, +}; + // One timer for every track so that a large session does not keep hundreds of // them, each firing a statistics request, in the event loop. const refreshStats$ = interval(1000).pipe(startWith(0), share()); @@ -72,17 +84,7 @@ export function createLivekitMediaTrack( room: LivekitRoom, ): AudioMediaTrack | VideoMediaTrack { const mediaChanged$ = observeParticipantMedia(participant); - const track$: Behavior = scope.behavior( - merge( - mediaChanged$, - fromEvent(publication, TrackEvent.Subscribed), - fromEvent(publication, TrackEvent.Unsubscribed), - ).pipe( - map(() => publication.track), - startWith(publication.track), - distinctUntilChanged(), - ), - ); + const track$ = publicationTrack$(scope, participant, publication); const attached = new Set(); let audioContext: AudioContext | undefined; @@ -172,13 +174,85 @@ export function createLivekitMediaTrack( }, }; + return { ...base, kind: "video" }; +} + +/** + * One of our own publications, with the controls over it. `setEnabled` goes + * through the local member so that what it remembers for a reconnection + * stays in step. + */ +export function createLocalLivekitMediaTrack( + scope: ObservableScope, + participant: LocalParticipant, + publication: TrackPublication, + room: LivekitRoom, + setEnabled: (source: MediaSource, enabled: boolean) => Promise, +): LocalAudioMediaTrack | LocalVideoMediaTrack { + const base = createLivekitMediaTrack(scope, participant, publication, room); + const local = { + setEnabled: async (enabled: boolean) => setEnabled(base.source, enabled), + setDevice: async (deviceId: string): Promise => { + if (base.source !== "microphone" && base.source !== "camera") + throw new MatrixRTCError(`A ${base.source} track has no device`); + await room.switchActiveDevice( + base.kind === "audio" ? "audioinput" : "videoinput", + deviceId, + ); + }, + }; + if (base.kind === "audio") return { ...base, ...local }; + + const videoTrack = (): LocalVideoTrack | undefined => + publication.track instanceof LocalVideoTrack + ? publication.track + : undefined; return { ...base, - kind: "video", - ...(participant.isLocal && { facingMode$: facingMode$(scope, track$) }), + ...local, + facingMode$: facingMode$( + scope, + publicationTrack$(scope, participant, publication), + ), + switchFacingMode: async () => { + const track = videoTrack(); + if (!track) return; + const { facingMode } = facingModeFromLocalTrack(track); + if (facingMode !== "user" && facingMode !== "environment") return; + await track.restartTrack({ + facingMode: facingMode === "user" ? "environment" : "user", + }); + return track.mediaStreamTrack.getSettings().deviceId; + }, + setProcessor: async (processor) => { + const track = videoTrack(); + // A processor cannot be built on a track that has already ended + if (!track || track.mediaStreamTrack.readyState === "ended") return; + if (processor) await track.setProcessor(processor); + else if (track.getProcessor()) await track.stopProcessor(); + }, }; } +/** The track behind a publication; a remote one arrives on subscription. */ +function publicationTrack$( + scope: ObservableScope, + participant: Participant, + publication: TrackPublication, +): Behavior { + return scope.behavior( + merge( + observeParticipantMedia(participant), + fromEvent(publication, TrackEvent.Subscribed), + fromEvent(publication, TrackEvent.Unsubscribed), + ).pipe( + map(() => publication.track), + startWith(publication.track), + distinctUntilChanged(), + ), + ); +} + async function rtpStreamStats( publication: TrackPublication, local: boolean, diff --git a/sdk/src/media/LivekitMemberMedia.ts b/sdk/src/media/LivekitMemberMedia.ts index b6a058a32..61fdf4e17 100644 --- a/sdk/src/media/LivekitMemberMedia.ts +++ b/sdk/src/media/LivekitMemberMedia.ts @@ -10,13 +10,11 @@ import { roomEventSelector, } from "@livekit/components-core"; import { - facingModeFromLocalTrack, type LocalParticipant, - LocalVideoTrack, type Participant, type Room as LivekitRoom, RoomEvent, - Track, + type TrackPublication, } from "livekit-client"; import { filter, map, type Observable } from "rxjs"; @@ -24,12 +22,17 @@ import { type ObservableScope } from "../reactive/ObservableScope"; import { generateItems } from "../reactive/observable"; import { E2eeType } from "../encryption"; import { type EncryptionSystem } from "../encryption"; +import { type Behavior } from "../reactive/Behavior"; import { type EncryptionError, type LocalMemberMedia, + type MediaSource, type MemberMedia, } from "../api"; -import { createLivekitMediaTrack } from "./LivekitMediaTrack"; +import { + createLivekitMediaTrack, + createLocalLivekitMediaTrack, +} from "./LivekitMediaTrack"; /** * A participant as `MemberMedia`. The only place, with `LivekitMediaTrack`, @@ -43,26 +46,9 @@ export function createLivekitMemberMedia( encryptionSystem: EncryptionSystem, ): MemberMedia { return { - local: participant.isLocal, - // One track per publication, each living as long as its publication is - // in the participant's map - tracks$: scope.behavior( - observeParticipantMedia(participant).pipe( - generateItems( - `${participant.identity} tracks$`, - function* () { - for (const publication of participant.trackPublications.values()) - yield { keys: [publication.trackSid], data: publication }; - }, - (trackScope, publication$) => - createLivekitMediaTrack( - trackScope, - participant, - publication$.value, - room, - ), - ), - ), + local: false, + tracks$: memberTracks$(scope, participant, (trackScope, publication) => + createLivekitMediaTrack(trackScope, participant, publication, room), ), encryptionError$: encryptionErrors$( scope, @@ -78,23 +64,48 @@ export function createLocalLivekitMemberMedia( participant: LocalParticipant, room: LivekitRoom, encryptionSystem: EncryptionSystem, + setEnabled: (source: MediaSource, enabled: boolean) => Promise, ): LocalMemberMedia { return { - ...createLivekitMemberMedia(scope, participant, room, encryptionSystem), 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; - }, + tracks$: memberTracks$(scope, participant, (trackScope, publication) => + createLocalLivekitMediaTrack( + trackScope, + participant, + publication, + room, + setEnabled, + ), + ), + encryptionError$: encryptionErrors$( + scope, + participant, + room, + encryptionSystem, + ), }; } +/** One track per publication, each living as long as its publication is in the participant's map. */ +function memberTracks$( + scope: ObservableScope, + participant: Participant, + factory: (scope: ObservableScope, publication: TrackPublication) => T, +): Behavior { + return scope.behavior( + observeParticipantMedia(participant).pipe( + generateItems( + `${participant.identity} tracks$`, + function* () { + for (const publication of participant.trackPublications.values()) + yield { keys: [publication.trackSid], data: publication }; + }, + (trackScope, publication$) => factory(trackScope, publication$.value), + ), + ), + ); +} + function encryptionErrors$( scope: ObservableScope, participant: Participant, diff --git a/sdk/src/session/ConnectionFactory.ts b/sdk/src/session/ConnectionFactory.ts index 347dd3fc6..4e4038d7b 100644 --- a/sdk/src/session/ConnectionFactory.ts +++ b/sdk/src/session/ConnectionFactory.ts @@ -138,8 +138,8 @@ function roomOptions( ...base, videoCaptureDefaults: { ...base.videoCaptureDefaults, - deviceId: localMedia.videoInputDeviceId$.value, - processor: localMedia.videoProcessor$.value, + deviceId: localMedia.videoInputDeviceId, + processor: localMedia.videoProcessor, ...(camera?.resolution && { resolution: camera.resolution }), }, publishDefaults: { @@ -154,7 +154,7 @@ function roomOptions( }, audioCaptureDefaults: { ...base.audioCaptureDefaults, - deviceId: localMedia.audioInputDeviceId$.value, + deviceId: localMedia.audioInputDeviceId, ...capture?.audio, }, audioOutput: { deviceId: localMedia.audioOutputDeviceId$.value }, diff --git a/sdk/src/session/LocalMember.ts b/sdk/src/session/LocalMember.ts index 2244c1b05..ce8dd6c23 100644 --- a/sdk/src/session/LocalMember.ts +++ b/sdk/src/session/LocalMember.ts @@ -46,7 +46,11 @@ import { type MatrixRTCMode, type SessionTimings, } from "../config"; -import { type DisconnectReason, type VideoCaptureSettings } from "../api"; +import { + type DisconnectReason, + type MediaSource, + type VideoCaptureSettings, +} from "../api"; import { FailToStartLivekitConnection, type MatrixRTCError, @@ -58,6 +62,7 @@ import { type Connection, ConnectionState } from "./Connection"; import { type IConnectionManager } from "./ConnectionManager"; import { type LocalTransport } from "./LocalTransport"; import { getSFUConfigWithOpenID } from "./openIDSFU"; +import { livekitSources } from "../media/LivekitMediaTrack"; import { type DesiredMedia, type Publisher } from "./Publisher"; export enum TransportState { @@ -118,6 +123,8 @@ export interface LocalMembership { disconnectReason$: Behavior; setMicrophoneEnabled: (enabled: boolean) => Promise; setCameraEnabled: (enabled: boolean) => Promise; + /** Mutes or unmutes one of our publications; the result is the state that resulted. */ + setEnabled: (source: MediaSource, enabled: boolean) => Promise; sharingScreen$: Behavior; toggleScreenSharing: (() => void) | null; screenShareError$: Behavior; @@ -392,22 +399,40 @@ export function createLocalMembership$({ }); const setEnabled = async ( - source: Track.Source.Microphone | Track.Source.Camera, + source: MediaSource, enabled: boolean, ): Promise => { + if (source !== "microphone" && source !== "camera") + return setPublicationEnabled(source, enabled); const desired$ = - source === Track.Source.Microphone - ? desired.microphone$ - : desired.camera$; + 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); + const result = await publisher.setEnabled( + source === "microphone" ? Track.Source.Microphone : Track.Source.Camera, + enabled, + ); if (result !== enabled) desired$.next(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 => { + 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))), @@ -451,10 +476,9 @@ export function createLocalMembership$({ connected$, reconnecting$, disconnectReason$, - setMicrophoneEnabled: async (enabled) => - setEnabled(Track.Source.Microphone, enabled), - setCameraEnabled: async (enabled) => - setEnabled(Track.Source.Camera, enabled), + setMicrophoneEnabled: async (enabled) => setEnabled("microphone", enabled), + setCameraEnabled: async (enabled) => setEnabled("camera", enabled), + setEnabled, sharingScreen$, toggleScreenSharing, screenShareError$, diff --git a/sdk/src/session/Members.ts b/sdk/src/session/Members.ts index 7b295e9aa..f2429ac61 100644 --- a/sdk/src/session/Members.ts +++ b/sdk/src/session/Members.ts @@ -93,6 +93,7 @@ export function createLocalRTCMember( participant, room, context.encryptionSystem, + localMembership.setEnabled, ), ), sharingScreen$: localMembership.sharingScreen$, diff --git a/sdk/src/session/Publisher.ts b/sdk/src/session/Publisher.ts index dc147ce05..e85c41312 100644 --- a/sdk/src/session/Publisher.ts +++ b/sdk/src/session/Publisher.ts @@ -5,24 +5,16 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -import { observeParticipantMedia } from "@livekit/components-core"; import { ConnectionState as LivekitConnectionState, type LocalTrackPublication, - LocalVideoTrack, ParticipantEvent, type Room as LivekitRoom, Track, } from "livekit-client"; import { type Logger } from "matrix-js-sdk/lib/logger"; -import { - type BehaviorSubject, - combineLatest, - distinctUntilChanged, - map, -} from "rxjs"; +import { type BehaviorSubject } from "rxjs"; -import { type Behavior } from "../reactive/Behavior"; import { ObservableScope } from "../reactive/ObservableScope"; import { type LocalMediaInputs } from "../api"; @@ -57,8 +49,7 @@ export class Publisher { room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => { this.logger.error("Failed to enable E2EE on the room", e); }); - this.followVideoProcessor(); - this.followDevices(); + this.followAudioOutput(); this.onLocalTrackPublished = this.onLocalTrackPublished.bind(this); room.localParticipant.on( ParticipantEvent.LocalTrackPublished, @@ -188,54 +179,19 @@ export class Publisher { } } - private followDevices(): void { - const sync = ( - kind: MediaDeviceKind, - deviceId$: Behavior, - ): void => { - deviceId$.pipe(this.scope.bind()).subscribe((deviceId) => { + private followAudioOutput(): void { + this.inputs.audioOutputDeviceId$ + .pipe(this.scope.bind()) + .subscribe((deviceId) => { if ( deviceId === undefined || this.room.state !== LivekitConnectionState.Connected || - this.room.getActiveDevice(kind) === deviceId + this.room.getActiveDevice("audiooutput") === deviceId ) return; this.room - .switchActiveDevice(kind, deviceId) - .catch((e) => this.logger.error(`Failed to switch ${kind}`, e)); - }); - }; - sync("audioinput", this.inputs.audioInputDeviceId$); - sync("audiooutput", this.inputs.audioOutputDeviceId$); - sync("videoinput", this.inputs.videoInputDeviceId$); - } - - private followVideoProcessor(): void { - const participant = this.room.localParticipant; - const cameraTrack$ = observeParticipantMedia(participant).pipe( - map(() => { - const track = participant.getTrackPublication( - Track.Source.Camera, - )?.track; - return track instanceof LocalVideoTrack ? track : undefined; - }), - distinctUntilChanged(), - ); - combineLatest([cameraTrack$, this.inputs.videoProcessor$]) - .pipe(this.scope.bind()) - .subscribe(([track, processor]) => { - if (!track) return; - if (processor && !track.getProcessor()) { - // A processor cannot be built on a track that has already ended - if (track.mediaStreamTrack.readyState === "ended") return; - track.setProcessor(processor).catch((e) => { - this.logger.warn("Failed to attach the video processor", e); - }); - } else if (!processor && track.getProcessor()) { - track.stopProcessor().catch((e) => { - this.logger.warn("Failed to stop the video processor", e); - }); - } + .switchActiveDevice("audiooutput", deviceId) + .catch((e) => this.logger.error("Failed to switch audiooutput", e)); }); } }