From 146dd3aea1678e06de7f952d27e4728ea142a430 Mon Sep 17 00:00:00 2001 From: fkwp Date: Wed, 4 Nov 2026 12:40:58 +0100 Subject: [PATCH] Start every camera track with the pipeline on A camera track made without the processor lets the room see the camera until the processor is built. So the processor travels in the publish request: the lobby opens the camera with it, the join publishes with it, the room's capture defaults follow the requests made since the join, and a camera turned off before the processor was chosen is replaced on turning it on rather than unmuted, so that the new track starts with it. Ported from spec/background-effects/2-pipeline (#4282): the publish half of a91dab8f. The replacement lives in the mute switch's track supply, which the call view model shares with the lobby. Co-Authored-By: Claude Fable 5.1 --- .../media-backend/livekit/Publisher.test.ts | 18 ++++ sdk/src/media-backend/livekit/Publisher.ts | 25 +++++- src/room/InCallView.tsx | 7 +- src/state/CallViewModel/CallViewModel.test.ts | 82 ++++++++++++++++++- src/state/CallViewModel/CallViewModel.ts | 50 +++++++++-- src/state/LobbyMediaViewModel.test.ts | 3 +- src/state/LobbyMediaViewModel.ts | 9 +- src/state/localMediaControls.ts | 17 +++- 8 files changed, 197 insertions(+), 14 deletions(-) diff --git a/sdk/src/media-backend/livekit/Publisher.test.ts b/sdk/src/media-backend/livekit/Publisher.test.ts index af595c65f..ffc8c2b51 100644 --- a/sdk/src/media-backend/livekit/Publisher.test.ts +++ b/sdk/src/media-backend/livekit/Publisher.test.ts @@ -120,6 +120,24 @@ describe("Publisher", () => { expect(participant.createTracks).not.toHaveBeenCalled(); }); + it("creates both with the options of requests made since the join", async () => { + const processor: VideoProcessor = { + name: "effect", + init: vi.fn(), + restart: vi.fn(), + destroy: vi.fn(), + }; + want(microphone, { source: "camera", deviceId: "front", processor }); + + publisher.start(); + await flushPromises(); + + expect(room.options.videoCaptureDefaults).toMatchObject({ + deviceId: "front", + processor: expect.objectContaining({ source: processor }), + }); + }); + it("asks for the camera and the microphone in one call", async () => { const enableBoth = vi.spyOn(participant, "enableCameraAndMicrophone"); want(microphone, camera); diff --git a/sdk/src/media-backend/livekit/Publisher.ts b/sdk/src/media-backend/livekit/Publisher.ts index 40728cef5..247b1256b 100644 --- a/sdk/src/media-backend/livekit/Publisher.ts +++ b/sdk/src/media-backend/livekit/Publisher.ts @@ -120,8 +120,10 @@ export class Publisher { // resolves these once the track is published, which may block on the // connection; LocalTrackPublished is what tells us a track exists. const captureBoth = toCapture.has("microphone") && toCapture.has("camera"); - if (captureBoth) + if (captureBoth) { + this.takeCaptureDefaults(requests); void this.room.localParticipant.enableCameraAndMicrophone(); + } for (const request of requests) if ( request.source === "screenShare" || @@ -133,6 +135,27 @@ export class Publisher { }); } + /** + * Makes the requests the room's capture defaults. The room took its defaults + * from the requests at the join, but one made since, such as a camera turned + * on before the start, has options of its own: a video processor dropped + * here would let the room see the camera before it is applied. + */ + private takeCaptureDefaults(requests: PublishRequest[]): void { + const options = this.room.options; + for (const request of requests) + if (request.source === "microphone") + options.audioCaptureDefaults = { + ...options.audioCaptureDefaults, + ...audioCaptureOptions(request), + }; + else if (request.source === "camera") + options.videoCaptureDefaults = { + ...options.videoCaptureDefaults, + ...videoCaptureOptions(request), + }; + } + private hasCaptured(source: MediaSource): boolean { return ( (source === "microphone" || source === "camera") && diff --git a/src/room/InCallView.tsx b/src/room/InCallView.tsx index 2ba8427a2..b6279b107 100644 --- a/src/room/InCallView.tsx +++ b/src/room/InCallView.tsx @@ -177,7 +177,12 @@ export const ActiveCall: FC = (props) => { publish: props.media.requests.length > 0 ? props.media - : initialPublishRequests(props.muteStates, mediaDevices, capture), + : initialPublishRequests( + props.muteStates, + mediaDevices, + capture, + backgroundEffectsState$.value.processor, + ), }); const vm = createCallViewModel$( diff --git a/src/state/CallViewModel/CallViewModel.test.ts b/src/state/CallViewModel/CallViewModel.test.ts index 33ffe0b73..4cf7ebfeb 100644 --- a/src/state/CallViewModel/CallViewModel.test.ts +++ b/src/state/CallViewModel/CallViewModel.test.ts @@ -28,6 +28,8 @@ import { type Layout } from "../layout-types.ts"; import { mockMatrixRoomMember, mockMediaDevices, + mockMuteStates, + flushPromises, mockRemoteParticipant, withTestScheduler, mockRtcMembership, @@ -62,7 +64,15 @@ import { import { MatrixRTCMode } from "../../config/ConfigOptions.ts"; import { initializeWidget } from "../../widget.ts"; import { computeUrlParams } from "../../UrlParams.ts"; -import { callViewModelOptionsFromParams } from "./CallViewModel.ts"; +import { + callViewModelOptionsFromParams, + initialPublishRequests, +} from "./CallViewModel.ts"; +import { type Handler, MuteState } from "../MuteStates.ts"; +import { + type RTCParticipation, + type VideoProcessor, +} from "@element-hq/matrixrtc-sdk"; initializeWidget(); @@ -1664,3 +1674,73 @@ describe("callViewModelOptionsFromParams", () => { expect(callViewModelOptionsFromParams(params).hideScreensharing).toBe(true); }); }); + +describe("a camera with a video processor", () => { + const withCallViewModel = withCallViewModelInMode(MatrixRTCMode.Matrix_2_0); + const processor = { name: "effect" } as VideoProcessor; + + /** The handler the view model gives the camera's mute switch. */ + const cameraHandler = (setHandler: { + mock: { calls: [Handler][] }; + }): Handler => setHandler.mock.calls[1][0]; + + it("starts each camera track with the processor", () => { + const muteStates = mockMuteStates(); + vi.spyOn(muteStates.video.enabled$, "value", "get").mockReturnValue(true); + expect( + initialPublishRequests( + muteStates, + mockMediaDevices({}), + {}, + processor, + ).find((request) => request.source === "camera"), + ).toMatchObject({ processor }); + }); + + it("replaces a camera turned off before the processor was chosen", async () => { + const setHandler = vi.spyOn(MuteState.prototype, "setHandler"); + onTestFinished(() => setHandler.mockRestore()); + const videoProcessor$ = new BehaviorSubject( + undefined, + ); + let participation: RTCParticipation | undefined; + withCallViewModel( + {}, + (_vm, _session, { rtcParticipation }) => { + participation = rtcParticipation; + }, + { videoProcessor$ }, + ); + // The camera is off, so it is stopped and can't take the processor + videoProcessor$.next(processor); + await cameraHandler(setHandler)(true); + await flushPromises(); + + expect(participation!.unpublish).toHaveBeenCalledWith("camera"); + expect(participation!.publish).toHaveBeenCalledWith( + expect.objectContaining({ source: "camera", processor }), + ); + }); + + it("turns on a camera that took the processor while running", async () => { + const setHandler = vi.spyOn(MuteState.prototype, "setHandler"); + onTestFinished(() => setHandler.mockRestore()); + const videoProcessor$ = new BehaviorSubject( + undefined, + ); + let participation: RTCParticipation | undefined; + withCallViewModel( + { videoEnabled: new Map([[localParticipant, constant(true)]]) }, + (_vm, _session, { rtcParticipation }) => { + participation = rtcParticipation; + }, + { videoProcessor$ }, + ); + videoProcessor$.next(processor); + await cameraHandler(setHandler)(true); + await flushPromises(); + + expect(participation!.unpublish).not.toHaveBeenCalled(); + expect(participation!.publish).not.toHaveBeenCalled(); + }); +}); diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index d7171aee9..42e310fe2 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -14,6 +14,7 @@ import { type EncryptionSystem, generateItems, type LocalAudioMediaTrack, + type LocalMediaTrack, type LocalRTCMember, type LocalVideoMediaTrack, type RTCParticipation, @@ -140,14 +141,16 @@ export interface CaptureSettings { } /** - * What to publish for a microphone or camera: the device the user picked and - * the host's capture settings. The video processor is not in here; the call - * attaches it to the camera track once that exists. + * What to publish for a microphone or camera: the device the user picked, the + * host's capture settings and, for the camera, the video processor, so that + * the track starts with it rather than letting the room through while it is + * built. */ export function publishRequest( source: "microphone" | "camera", mediaDevices: MediaDevices, capture: CaptureSettings, + processor?: VideoProcessor, ): PublishRequest { return source === "microphone" ? { @@ -159,6 +162,7 @@ export function publishRequest( source, deviceId: mediaDevices.videoInput.selected$.value?.id, capture: capture.camera, + ...(processor && { processor }), }; } @@ -167,12 +171,13 @@ export function initialPublishRequests( muteStates: MuteStates, mediaDevices: MediaDevices, capture: CaptureSettings, + processor?: VideoProcessor, ): PublishRequest[] { const requests: PublishRequest[] = []; if (muteStates.audio.enabled$.value) requests.push(publishRequest("microphone", mediaDevices, capture)); if (muteStates.video.enabled$.value) - requests.push(publishRequest("camera", mediaDevices, capture)); + requests.push(publishRequest("camera", mediaDevices, capture, processor)); return requests; } @@ -508,9 +513,36 @@ export function createCallViewModel$( "screenShare", ) as Behavior; + // Camera tracks the processor reached while they ran. A camera turned off is + // stopped, and a stopped track can't take one. + const tookProcessor = new WeakMap(); + const missesProcessor = (track: LocalMediaTrack): boolean => { + const processor = videoProcessor$.value; + return ( + processor !== undefined && + tookProcessor.get(track as LocalVideoMediaTrack) !== processor + ); + }; const supply = (source: "microphone" | "camera"): TrackSupply => ({ publish: async () => - rtcParticipation.publish(publishRequest(source, mediaDevices, capture)), + rtcParticipation.publish( + publishRequest( + source, + mediaDevices, + capture, + source === "camera" ? videoProcessor$.value : undefined, + ), + ), + // A camera turned off before the processor was chosen is replaced, so + // that the new one starts with it + replace: + source === "camera" + ? async (track): Promise => { + if (!missesProcessor(track)) return false; + await rtcParticipation.unpublish("camera"); + return true; + } + : undefined, }); followMuteState( scope, @@ -550,7 +582,13 @@ export function createCallViewModel$( }); }); - followProcessor(scope, camera$, videoProcessor$, logger); + followProcessor( + scope, + camera$, + videoProcessor$, + logger, + (camera, processor) => tookProcessor.set(camera, processor), + ); const members$ = scope.behavior<(LocalRTCMember | RemoteRTCMember)[]>( combineLatest( diff --git a/src/state/LobbyMediaViewModel.test.ts b/src/state/LobbyMediaViewModel.test.ts index 7b2bc2b7b..5be598497 100644 --- a/src/state/LobbyMediaViewModel.test.ts +++ b/src/state/LobbyMediaViewModel.test.ts @@ -29,7 +29,8 @@ describe("LobbyMediaViewModel", () => { expect(media.capture).toHaveBeenCalledExactlyOnceWith([ expect.objectContaining({ source: "microphone", deviceId: "mic" }), - expect.objectContaining({ source: "camera", deviceId: "cam" }), + // With the processor, so that the camera starts with it + expect.objectContaining({ source: "camera", deviceId: "cam", processor }), ]); const camera = vm.videoTrack$.value!; expect(camera.source).toBe("camera"); diff --git a/src/state/LobbyMediaViewModel.ts b/src/state/LobbyMediaViewModel.ts index 9a2061d1f..b955fb63c 100644 --- a/src/state/LobbyMediaViewModel.ts +++ b/src/state/LobbyMediaViewModel.ts @@ -71,7 +71,14 @@ export function createLobbyMediaViewModel( if (audio && microphone === undefined && !opening.has("microphone")) open.push(publishRequest("microphone", mediaDevices, capture)); if (video && camera === undefined && !opening.has("camera")) - open.push(publishRequest("camera", mediaDevices, capture)); + open.push( + publishRequest( + "camera", + mediaDevices, + capture, + videoProcessor$.value, + ), + ); if (open.length > 0) { for (const { source } of open) opening.add(source); media diff --git a/src/state/localMediaControls.ts b/src/state/localMediaControls.ts index 32efe3a7d..61521f988 100644 --- a/src/state/localMediaControls.ts +++ b/src/state/localMediaControls.ts @@ -18,9 +18,14 @@ import { combineLatest } from "rxjs"; import { type MediaDevice } from "./MediaDevices"; import { type Handler } from "./MuteStates"; -/** How a mute switch gets a track where there is none. */ +/** + * How a mute switch gets a track where there is none and, where given, + * whether a track it has should be let go and published anew instead of + * unmuted: resolves true once it is gone. + */ export interface TrackSupply { publish: () => Promise; + replace?: (track: LocalMediaTrack) => Promise; } /** @@ -46,7 +51,9 @@ export function followMuteState( muteState.setHandler(async (enabled) => { desired = enabled; const track = track$.value; - if (track !== undefined) { + const replaced = + enabled && track !== undefined && (await supply.replace?.(track)); + if (track !== undefined && !replaced) { desired = await track.setEnabled(enabled); return desired; } @@ -100,11 +107,15 @@ export function followProcessor( camera$: Behavior, videoProcessor$: Behavior, logger: Logger, + /** Told of each camera a processor reaches while it runs. */ + applied?: (camera: LocalVideoMediaTrack, processor: VideoProcessor) => void, ): void { combineLatest([camera$, videoProcessor$]) .pipe(scope.bind()) .subscribe(([camera, processor]) => { - camera?.setProcessor(processor).catch((e) => { + if (camera === undefined) return; + if (processor && !camera.muted$.value) applied?.(camera, processor); + camera.setProcessor(processor).catch((e) => { logger.error("Failed to set the video processor", e); }); });