diff --git a/sdk/src/media-backend/livekit/LivekitMediaTrack.ts b/sdk/src/media-backend/livekit/LivekitMediaTrack.ts index dd1b81c93..3b07d2455 100644 --- a/sdk/src/media-backend/livekit/LivekitMediaTrack.ts +++ b/sdk/src/media-backend/livekit/LivekitMediaTrack.ts @@ -230,9 +230,10 @@ export function createLocalLivekitMediaTrack( 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(convertToLivekitProcessor(processor)); - else if (track.getProcessor()) await track.stopProcessor(); + if (processor) { + const adapter = convertToLivekitProcessor(processor); + if (track.getProcessor() !== adapter) await track.setProcessor(adapter); + } else if (track.getProcessor()) await track.stopProcessor(); }, }; } diff --git a/sdk/src/media-backend/livekit/Publisher.test.ts b/sdk/src/media-backend/livekit/Publisher.test.ts index 840ae3ad6..80ffbede2 100644 --- a/sdk/src/media-backend/livekit/Publisher.test.ts +++ b/sdk/src/media-backend/livekit/Publisher.test.ts @@ -18,9 +18,10 @@ import { import { logger } from "matrix-js-sdk/lib/logger"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { type PublishRequest } from "../../media-api"; +import { type PublishRequest, type VideoProcessor } from "../../media-api"; import { flushPromises } from "../../utils/test"; import { type DesiredMedia, Publisher } from "./Publisher"; +import { convertToLivekitProcessor } from "./videoProcessor"; describe("Publisher", () => { describe("setAudioOutputDevice", () => { @@ -119,6 +120,24 @@ describe("Publisher", () => { }); }); + 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: convertToLivekitProcessor(processor), + }); + }); + it.each([ { enabled: microphone, muted: camera, options: { audio: true } }, { enabled: camera, muted: microphone, options: { video: true } }, diff --git a/sdk/src/media-backend/livekit/Publisher.ts b/sdk/src/media-backend/livekit/Publisher.ts index 00681644b..9c4653b35 100644 --- a/sdk/src/media-backend/livekit/Publisher.ts +++ b/sdk/src/media-backend/livekit/Publisher.ts @@ -87,8 +87,10 @@ export class Publisher { // 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")) + if (sources.has("microphone") && sources.has("camera")) { + this.takeCaptureDefaults(requests); void this.room.localParticipant.enableCameraAndMicrophone(); + } for (const request of requests) if ( request.source === "screenShare" || @@ -99,6 +101,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), + }; + } + /** Publishes one request and resolves with its publication once LiveKit has it. */ public async publish( request: PublishRequest, diff --git a/sdk/src/media-backend/livekit/videoProcessor.test.ts b/sdk/src/media-backend/livekit/videoProcessor.test.ts index 468379fe0..b9d26e414 100644 --- a/sdk/src/media-backend/livekit/videoProcessor.test.ts +++ b/sdk/src/media-backend/livekit/videoProcessor.test.ts @@ -47,4 +47,11 @@ describe("convertToLivekitProcessor", () => { await adapter.destroy(); expect(processor.destroy).toHaveBeenCalled(); }); + + it("gives one processor one adapter", () => { + const processor = fakeProcessor(); + expect(convertToLivekitProcessor(processor)).toBe( + convertToLivekitProcessor(processor), + ); + }); }); diff --git a/sdk/src/media-backend/livekit/videoProcessor.ts b/sdk/src/media-backend/livekit/videoProcessor.ts index 546ea2113..fdf84d4e3 100644 --- a/sdk/src/media-backend/livekit/videoProcessor.ts +++ b/sdk/src/media-backend/livekit/videoProcessor.ts @@ -9,14 +9,24 @@ import { type Track, type TrackProcessor } from "livekit-client"; import { type VideoProcessor } from "../../media-api"; +const adapters = new WeakMap< + VideoProcessor, + TrackProcessor +>(); + /** * A `VideoProcessor` as livekit-client wants it. LiveKit reads the processed * track off the processor after `init` and `restart` rather than taking it as * their result, so the adapter keeps it there. + * + * One adapter per processor, so a track can tell that it already runs one: a + * processor set again would otherwise be torn down and built anew. */ export function convertToLivekitProcessor( processor: VideoProcessor, ): TrackProcessor { + const existing = adapters.get(processor); + if (existing) return existing; const adapter: TrackProcessor = { name: processor.name, init: async (opts) => { @@ -30,5 +40,6 @@ export function convertToLivekitProcessor( }, destroy: async () => processor.destroy(), }; + adapters.set(processor, adapter); return adapter; } diff --git a/src/room/InCallView.tsx b/src/room/InCallView.tsx index 71d32a406..9caba2841 100644 --- a/src/room/InCallView.tsx +++ b/src/room/InCallView.tsx @@ -170,7 +170,12 @@ export const ActiveCall: FC = (props) => { ); const rtcParticipation = slot.join({ ...rtcParticipationOptions(urlParams), - publish: initialPublishRequests(props.muteStates, mediaDevices, capture), + publish: initialPublishRequests( + props.muteStates, + mediaDevices, + capture, + trackProcessorState$.value.processor, + ), }); const vm = createCallViewModel$( diff --git a/src/state/BackgroundEffects.test.ts b/src/state/BackgroundEffects.test.ts new file mode 100644 index 000000000..7319f11b5 --- /dev/null +++ b/src/state/BackgroundEffects.test.ts @@ -0,0 +1,131 @@ +/* +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 { beforeEach, describe, expect, it, vi } from "vitest"; +import { BehaviorSubject, distinctUntilChanged, map } from "rxjs"; +import { type BackgroundProcessorWrapper } from "@livekit/track-processors"; + +import { + BackgroundEffects, + type BackgroundEffectsOptions, +} from "./BackgroundEffects"; +import { type ProcessorState } from "../tracks/TrackProcessorContext"; +import { flushPromises, testScope, withTestScheduler } from "../utils/test"; + +/** A pipeline that records what it is switched to. */ +function fakePipeline(): { + pipeline: BackgroundProcessorWrapper; + switches: unknown[]; + /** Holds the next switch until it resolves, as a picture loading does. */ + holdNext: () => () => void; +} { + const switches: unknown[] = []; + let slow: Promise | undefined; + const pipeline = { + switchTo: vi.fn(async (options: unknown) => { + switches.push(options); + const held = slow; + slow = undefined; + await held; + }), + } as unknown as BackgroundProcessorWrapper; + return { + pipeline, + switches, + holdNext: () => { + let release!: () => void; + slow = new Promise((resolve) => (release = resolve)); + return release; + }, + }; +} + +/** One letter per state: idle, attached. */ +function letter(state: ProcessorState): string { + return state.processor === undefined ? "i" : "a"; +} + +describe("the pipeline's state", () => { + function testState({ + effect, + expected, + }: { + effect: string; + expected: string; + }): void { + withTestScheduler(({ behavior, expectObservable }) => { + const effects = new BackgroundEffects(testScope(), { + supported: true, + blur$: behavior(effect, { n: false, b: true }), + pipeline: fakePipeline().pipeline, + }); + expectObservable( + effects.state$.pipe(map(letter), distinctUntilChanged()), + ).toBe(expected); + }); + } + + it("attaches on first use and stays attached", () => + testState({ effect: "nbn", expected: "ia" })); +}); + +describe("background effects", () => { + let blur$: BehaviorSubject; + let fake: ReturnType; + + function build( + options: Partial = {}, + ): BackgroundEffects { + return new BackgroundEffects(testScope(), { + supported: true, + blur$, + pipeline: fake.pipeline, + ...options, + }); + } + const blur = async (on: boolean): Promise => { + blur$.next(on); + await flushPromises(); + }; + + beforeEach(() => { + blur$ = new BehaviorSubject(false); + fake = fakePipeline(); + }); + + it("switches in place rather than reattaching", async () => { + const effects = build(); + await blur(true); + const pipeline = effects.state$.value.processor; + await blur(false); + await blur(true); + + expect(effects.state$.value.processor).toBe(pipeline); + expect(fake.switches).toEqual([ + { mode: "background-blur", blurRadius: 15 }, + { mode: "disabled" }, + { mode: "background-blur", blurRadius: 15 }, + ]); + }); + + it("switches one at a time, skipping those overtaken", async () => { + const finished = fake.holdNext(); + build(); + await blur(true); + await blur(false); + await blur(true); + await blur(false); + expect(fake.switches).toHaveLength(1); + + finished(); + await flushPromises(); + expect(fake.switches).toEqual([ + { mode: "background-blur", blurRadius: 15 }, + { mode: "disabled" }, + ]); + }); +}); diff --git a/src/state/BackgroundEffects.ts b/src/state/BackgroundEffects.ts new file mode 100644 index 000000000..e89939805 --- /dev/null +++ b/src/state/BackgroundEffects.ts @@ -0,0 +1,98 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { combineLatest, distinctUntilChanged, filter, map, scan } from "rxjs"; +import { + type BackgroundProcessorWrapper, + type SwitchBackgroundProcessorOptions, +} from "@livekit/track-processors"; +import { logger } from "matrix-js-sdk/lib/logger"; + +import { type Behavior } from "./Behavior"; +import { type ObservableScope } from "./ObservableScope"; +import { + fromLivekitProcessor, + type ProcessorState, +} from "../tracks/TrackProcessorContext"; + +const blurRadius = 15; + +export interface BackgroundEffectsOptions { + /** Whether this browser can run a pipeline at all. */ + supported: boolean; + /** Whether blur is chosen. */ + blur$: Behavior; + /** + * Shared by the pre-join preview and the call. Building or destroying it is + * what primes it, and a primed pipeline lets its next frame through + * untouched, so it is switched rather than rebuilt. + */ + pipeline: BackgroundProcessorWrapper; +} + +/** The background effect pipeline, as the camera tracks and the menus see it. */ +export class BackgroundEffects { + public readonly state$: Behavior; + + public constructor( + scope: ObservableScope, + { supported, blur$, pipeline }: BackgroundEffectsOptions, + ) { + // Wrapped once, so the call and the preview see one processor throughout. + const processor = fromLivekitProcessor(pipeline); + this.state$ = scope.behavior( + blur$.pipe( + scan( + (previous, wanted) => { + // Attached the first time an effect is wanted and never detached + // after, so someone who never turns one on pays for none of it. + const attached = + previous.processor !== undefined || (supported && wanted); + return { supported, processor: attached ? processor : undefined }; + }, + { supported, processor: undefined }, + ), + ), + ); + + const switchTo = oneSwitchAtATime(pipeline); + combineLatest([this.state$, blur$]) + .pipe( + filter(([{ processor }]) => processor !== undefined), + map(([, blur]): SwitchBackgroundProcessorOptions => + blur ? { mode: "background-blur", blurRadius } : { mode: "disabled" }, + ), + distinctUntilChanged((a, b) => JSON.stringify(a) === JSON.stringify(b)), + scope.bind(), + ) + .subscribe((options) => { + switchTo(options).catch((e) => + logger.warn("Failed to switch background effect", e), + ); + }); + } +} + +/** + * Switches the pipeline one choice at a time, skipping those overtaken while + * they waited. A switch to a picture ends only once it has loaded, so a + * slower, earlier choice would otherwise land after a later one. + */ +function oneSwitchAtATime( + pipeline: BackgroundProcessorWrapper, +): (options: SwitchBackgroundProcessorOptions) => Promise { + let latest: SwitchBackgroundProcessorOptions | undefined; + let queue = Promise.resolve(); + return async (options) => { + latest = options; + const turn = queue.then(async () => { + if (options === latest) await pipeline.switchTo(options); + }); + queue = turn.catch(() => {}); + return turn; + }; +} diff --git a/src/state/CallViewModel/CallViewModel.test.ts b/src/state/CallViewModel/CallViewModel.test.ts index 33ffe0b73..82c98741b 100644 --- a/src/state/CallViewModel/CallViewModel.test.ts +++ b/src/state/CallViewModel/CallViewModel.test.ts @@ -23,6 +23,10 @@ import { MatrixRTCSessionEvent, } from "matrix-js-sdk/lib/matrixrtc"; import { deepCompare } from "matrix-js-sdk/lib/utils"; +import { + type RTCParticipation, + type VideoProcessor, +} from "@element-hq/matrixrtc-sdk"; import { type Layout } from "../layout-types.ts"; import { @@ -31,7 +35,9 @@ import { mockRemoteParticipant, withTestScheduler, mockRtcMembership, + mockMuteStates, testScope, + flushPromises, } from "../../utils/test.ts"; import { E2eeType } from "../../e2ee/e2eeType.ts"; import { @@ -62,7 +68,11 @@ 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"; 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 ea777b12f..797a8da09 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -135,14 +135,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" ? { @@ -154,6 +156,7 @@ export function publishRequest( source, deviceId: mediaDevices.videoInput.selected$.value?.id, capture: capture.camera, + ...(processor && { processor }), }; } @@ -162,12 +165,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; } @@ -503,6 +507,17 @@ 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 + ); + }; + // The mute switches drive our tracks, which report back what the device // allowed, so a denied permission flips the switch back. A source is muted // and unmuted while its track exists and published when it does not. @@ -520,14 +535,23 @@ export function createCallViewModel$( muteState.setHandler(async (enabled) => { desired = enabled; const track = track$.value; - if (track !== undefined) { + // A camera turned off before the processor was chosen is replaced, so + // that the new one starts with it. + if (enabled && source === "camera" && track && missesProcessor(track)) + await rtcParticipation.unpublish("camera"); + else if (track !== undefined) { desired = await track.setEnabled(enabled); return desired; } if (!enabled) return false; try { await rtcParticipation.publish( - publishRequest(source, mediaDevices, capture), + publishRequest( + source, + mediaDevices, + capture, + source === "camera" ? videoProcessor$.value : undefined, + ), ); return true; } catch (e) { @@ -583,12 +607,15 @@ export function createCallViewModel$( }); }); - // Attached once the camera track exists rather than carried in the publish - // request, so that a new track and a change of processor take the same path + // Also set on each camera track as it appears and as the choice changes. A + // track made with the processor already runs it and is left alone. combineLatest([camera$, videoProcessor$]) .pipe(scope.bind()) .subscribe(([camera, processor]) => { - camera?.setProcessor(processor).catch((e) => { + if (camera === undefined) return; + if (processor && !camera.muted$.value) + tookProcessor.set(camera, processor); + camera.setProcessor(processor).catch((e) => { logger.error("Failed to set the video processor", e); }); }); diff --git a/src/tracks/BackgroundEffectTransformer.test.ts b/src/tracks/BackgroundEffectTransformer.test.ts new file mode 100644 index 000000000..b8a26b86b --- /dev/null +++ b/src/tracks/BackgroundEffectTransformer.test.ts @@ -0,0 +1,24 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { describe, expect, it } from "vitest"; + +import { BackgroundEffectTransformer } from "./BackgroundEffectTransformer"; + +describe("BackgroundEffectTransformer", () => { + it("passes frames on untouched once disabled, even after blurring", async () => { + const transformer = new BackgroundEffectTransformer({}); + await transformer.update({ blurRadius: 15, backgroundDisabled: false }); + await transformer.update({ + imagePath: undefined, + backgroundDisabled: true, + }); + // With neither, the library passes a frame on without segmenting it. + expect(transformer.options.blurRadius).toBeUndefined(); + expect(transformer.options.imagePath).toBeUndefined(); + }); +}); diff --git a/src/tracks/BlurBackgroundTransformer.ts b/src/tracks/BackgroundEffectTransformer.ts similarity index 78% rename from src/tracks/BlurBackgroundTransformer.ts rename to src/tracks/BackgroundEffectTransformer.ts index f86120d33..965359211 100644 --- a/src/tracks/BlurBackgroundTransformer.ts +++ b/src/tracks/BackgroundEffectTransformer.ts @@ -1,5 +1,6 @@ /* Copyright 2024-2025 New Vector Ltd. +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. @@ -9,6 +10,7 @@ import { BackgroundTransformer, VideoTransformer, type VideoTransformerInitOptions, + type BackgroundOptions, } from "@livekit/track-processors"; import { ImageSegmenter } from "@mediapipe/tasks-vision"; @@ -49,7 +51,18 @@ const wasmFileset: WasmFileset = { * loads the segmentation models from our own bundle rather than as an external * resource fetched from the public internet. */ -export class BlurBackgroundTransformer extends BackgroundTransformer { +export class BackgroundEffectTransformer extends BackgroundTransformer { + /** + * As the library's, except that disabling also drops the blur radius: kept, + * it has every frame segmented and the result thrown away, where with + * neither a radius nor a picture frames are passed on untouched. + */ + public override async update(opts: BackgroundOptions): Promise { + await super.update( + opts.backgroundDisabled ? { ...opts, blurRadius: undefined } : opts, + ); + } + public async init({ outputCanvas, inputElement: inputVideo, @@ -73,8 +86,10 @@ export class BlurBackgroundTransformer extends BackgroundTransformer { outputConfidenceMasks: false, }); + // BackgroundTransformer's own init applies these, and this one replaces it. if (this.options.blurRadius) { this.gl?.setBlurRadius(this.options.blurRadius); } + this.gl?.setBackgroundDisabled(this.options.backgroundDisabled ?? false); } } diff --git a/src/tracks/OneStepPipeline.test.ts b/src/tracks/OneStepPipeline.test.ts new file mode 100644 index 000000000..12aa94773 --- /dev/null +++ b/src/tracks/OneStepPipeline.test.ts @@ -0,0 +1,80 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ProcessorWrapper } from "@livekit/track-processors"; + +import { OneStepPipeline } from "./OneStepPipeline"; +import { BackgroundEffectTransformer } from "./BackgroundEffectTransformer"; +import { flushPromises } from "../utils/test"; + +const opts = {} as Parameters[0]; + +describe("OneStepPipeline", () => { + afterEach(() => vi.restoreAllMocks()); + + it("stops and builds again only once a build has finished", async () => { + let finishBuild!: () => void; + const init = vi + .spyOn(ProcessorWrapper.prototype, "init") + .mockImplementationOnce( + async () => new Promise((resolve) => (finishBuild = resolve)), + ) + .mockResolvedValue(); + const destroy = vi + .spyOn(ProcessorWrapper.prototype, "destroy") + .mockResolvedValue(); + const pipeline = new OneStepPipeline( + new BackgroundEffectTransformer({}), + "test", + ); + + const first = pipeline.init(opts); + const stopped = pipeline.destroy(); + const second = pipeline.init(opts); + await flushPromises(); + expect(init).toHaveBeenCalledTimes(1); + expect(destroy).not.toHaveBeenCalled(); + + finishBuild(); + await Promise.all([first, stopped, second]); + expect(init).toHaveBeenCalledTimes(2); + expect(destroy.mock.invocationCallOrder[0]).toBeLessThan( + init.mock.invocationCallOrder[1], + ); + }); + + it("restarts in one step, without waiting on itself", async () => { + const init = vi + .spyOn(ProcessorWrapper.prototype, "init") + .mockResolvedValue(); + const destroy = vi + .spyOn(ProcessorWrapper.prototype, "destroy") + .mockResolvedValue(); + const pipeline = new OneStepPipeline( + new BackgroundEffectTransformer({}), + "test", + ); + + await pipeline.restart(opts); + expect(destroy).toHaveBeenCalledWith({ willProcessorRestart: true }); + expect(init).toHaveBeenCalledOnce(); + }); + + it("carries on after a step that failed", async () => { + vi.spyOn(ProcessorWrapper.prototype, "init") + .mockRejectedValueOnce(new Error("no GPU")) + .mockResolvedValue(); + const pipeline = new OneStepPipeline( + new BackgroundEffectTransformer({}), + "test", + ); + + await expect(pipeline.init(opts)).rejects.toThrow("no GPU"); + await expect(pipeline.init(opts)).resolves.toBeUndefined(); + }); +}); diff --git a/src/tracks/OneStepPipeline.ts b/src/tracks/OneStepPipeline.ts new file mode 100644 index 000000000..e2060a0ea --- /dev/null +++ b/src/tracks/OneStepPipeline.ts @@ -0,0 +1,43 @@ +/* +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 { BackgroundProcessorWrapper } from "@livekit/track-processors"; + +type InitOptions = Parameters[0]; +type DestroyOptions = Parameters[0]; + +/** + * A background pipeline that is built, rebuilt and destroyed one step at a + * time. The camera tracks sharing it attach and stop it independently, and + * the SDK's wrapper holds a single track's streams: two builds at once + * overwrite each other's, and a destroy during a build does nothing. + */ +export class OneStepPipeline extends BackgroundProcessorWrapper { + private steps: Promise = Promise.resolve(); + + public override async init(opts: InitOptions): Promise { + return this.inTurn(async () => super.init(opts)); + } + + // The SDK's restart calls init and destroy, which would wait on itself. + public override async restart(opts: InitOptions): Promise { + return this.inTurn(async () => { + await super.destroy({ willProcessorRestart: true }); + await super.init(opts); + }); + } + + public override async destroy(options?: DestroyOptions): Promise { + return this.inTurn(async () => super.destroy(options)); + } + + private async inTurn(step: () => Promise): Promise { + const turn = this.steps.then(step); + this.steps = turn.catch(() => {}); + return turn; + } +} diff --git a/src/tracks/TrackProcessorContext.test.ts b/src/tracks/TrackProcessorContext.test.ts index 5206a5837..fc8966030 100644 --- a/src/tracks/TrackProcessorContext.test.ts +++ b/src/tracks/TrackProcessorContext.test.ts @@ -5,22 +5,59 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -import { describe, expect, it, vi } from "vitest"; +import { act, createElement, type FC } from "react"; +import { render } from "@testing-library/react"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { type LocalVideoTrack } from "livekit-client"; import { type BackgroundOptions, type ProcessorWrapper, } from "@livekit/track-processors"; -import { type VideoProcessor } from "@element-hq/matrixrtc-sdk"; +import { + convertToLivekitProcessor, + type VideoProcessor, +} from "@element-hq/matrixrtc-sdk"; import { applyProcessor, fromLivekitProcessor, + ProcessorProvider, + type ProcessorState, trackProcessorSync, + useTrackProcessor, + useTrackProcessorSync, } from "./TrackProcessorContext"; +import { backgroundBlur } from "../settings/settings"; import { constant } from "../state/Behavior"; import { flushPromises, testScope } from "../utils/test"; +const pipelines = vi.hoisted(() => ({ + built: 0, + destroyed: 0, + switches: [] as unknown[], +})); + +// The pipeline itself needs WebGL and MediaPipe; what these tests are about is +// what the provider asks of it. +vi.mock("@livekit/track-processors", () => ({ + BackgroundProcessorWrapper: vi.fn(function (this: Record) { + pipelines.built += 1; + this.switchTo = vi.fn(async (options: unknown) => { + pipelines.switches.push(options); + return Promise.resolve(); + }); + this.destroy = vi.fn(async () => { + pipelines.destroyed += 1; + return Promise.resolve(); + }); + }), + supportsBackgroundProcessors: (): boolean => true, +})); +vi.mock("./BackgroundEffectTransformer", () => ({ + BackgroundEffectTransformer: vi.fn(), +})); +vi.mock("../Platform", () => ({ platform: "desktop" })); + const processor = { name: "blur" } as VideoProcessor; function mockTrack( @@ -131,3 +168,111 @@ describe("fromLivekitProcessor", () => { ).rejects.toThrow("produced no track"); }); }); + +describe("ProcessorProvider", () => { + /** A camera track that holds whatever processor it is given. */ + function cameraTrack(): LocalVideoTrack { + let current: unknown; + return { + mediaStreamTrack: { readyState: "live" }, + getProcessor: vi.fn(() => current), + setProcessor: vi.fn(async (next: unknown) => { + current = next; + return Promise.resolve(); + }), + stopProcessor: vi.fn(async () => { + current = undefined; + return Promise.resolve(); + }), + } as unknown as LocalVideoTrack; + } + + let seen: ProcessorState[]; + const Surface: FC<{ track: LocalVideoTrack | null }> = ({ track }) => { + seen.push(useTrackProcessor()); + useTrackProcessorSync(track); + return null; + }; + // One component for every render, so a rerender updates the tree rather than + // mounting a second provider with a pipeline of its own. + const Surfaces: FC<{ tracks: (LocalVideoTrack | null)[] }> = ({ tracks }) => + createElement( + ProcessorProvider, + null, + createElement( + "div", + null, + ...tracks.map((track, key) => createElement(Surface, { key, track })), + ), + ); + const surfaces = ( + ...tracks: (LocalVideoTrack | null)[] + ): ReturnType => createElement(Surfaces, { tracks }); + const blur = async (on: boolean): Promise => { + await act(async () => { + backgroundBlur.setValue(on); + await flushPromises(); + }); + }; + const latest = (): ProcessorState => seen[seen.length - 1]; + + beforeEach(() => { + seen = []; + pipelines.built = 0; + pipelines.destroyed = 0; + pipelines.switches = []; + backgroundBlur.setValue(false); + }); + afterEach(() => backgroundBlur.setValue(false)); + + // A pipeline primes itself when it is built and when it is destroyed, and a + // primed pipeline lets one frame through untouched: so the frame is spent + // once, as long as it is built once and never taken off the camera. + it("spends one priming frame on first use and none afterwards", async () => { + const track = cameraTrack(); + render(surfaces(track)); + for (const on of [true, false, true, false, true]) await blur(on); + + expect(pipelines.built).toBe(1); + expect(pipelines.destroyed).toBe(0); + expect(track.setProcessor).toHaveBeenCalledTimes(1); + expect(track.stopProcessor).not.toHaveBeenCalled(); + }); + + it("preview and call share one pipeline", async () => { + const preview = cameraTrack(); + const call = cameraTrack(); + render(surfaces(preview, call)); + await blur(true); + + const pipeline = latest().processor; + expect(preview.setProcessor).toHaveBeenCalledWith( + convertToLivekitProcessor(pipeline!), + ); + expect(call.setProcessor).toHaveBeenCalledWith( + convertToLivekitProcessor(pipeline!), + ); + expect(pipelines.built).toBe(1); + }); + + it("applies an effect chosen while the camera is off", async () => { + const view = render(surfaces(null)); + await blur(true); + // Already switched to blur, with nothing yet to attach it to. + expect(pipelines.switches).toEqual([ + { mode: "background-blur", blurRadius: 15 }, + ]); + + const track = cameraTrack(); + await act(async () => { + view.rerender(surfaces(track)); + await flushPromises(); + }); + expect(track.setProcessor).toHaveBeenCalledTimes(1); + expect(track.setProcessor).toHaveBeenCalledWith( + convertToLivekitProcessor(latest().processor!), + ); + expect(pipelines.built).toBe(1); + expect(pipelines.switches).toHaveLength(1); + }); +}); diff --git a/src/tracks/TrackProcessorContext.tsx b/src/tracks/TrackProcessorContext.tsx index 785928b62..85ffc3aa3 100644 --- a/src/tracks/TrackProcessorContext.tsx +++ b/src/tracks/TrackProcessorContext.tsx @@ -1,12 +1,13 @@ /* Copyright 2024-2025 New Vector Ltd. +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 { - ProcessorWrapper, + type ProcessorWrapper, supportsBackgroundProcessors as supportsBackgroundProcessorsLivekitSdk, type BackgroundOptions, } from "@livekit/track-processors"; @@ -16,7 +17,7 @@ import { type JSX, use, useEffect, - useMemo, + useState, } from "react"; import { type LocalVideoTrack, Track } from "livekit-client"; import { logger } from "matrix-js-sdk/lib/logger"; @@ -26,15 +27,14 @@ import { type VideoProcessor, } from "@element-hq/matrixrtc-sdk"; -import { - backgroundBlur as backgroundBlurSettings, - useSetting, -} from "../settings/settings"; -import { BlurBackgroundTransformer } from "./BlurBackgroundTransformer"; +import { backgroundBlur as backgroundBlurSettings } from "../settings/settings"; +import { BackgroundEffectTransformer } from "./BackgroundEffectTransformer"; +import { OneStepPipeline } from "./OneStepPipeline"; import { type Behavior } from "../state/Behavior"; -import { type ObservableScope } from "../state/ObservableScope"; +import { ObservableScope } from "../state/ObservableScope"; +import { BackgroundEffects } from "../state/BackgroundEffects"; +import { useBehavior } from "../useBehavior"; import { platform } from "../Platform"; -import { useValueBehavior } from "../useValueBehavior"; //TODO-MULTI-SFU: This is not yet fully there. // it is a combination of exposing observable and react hooks. @@ -72,19 +72,21 @@ export function fromLivekitProcessor( }; } -const ProcessorContext = createContext(undefined); +const ProcessorContext = createContext( + undefined, +); export function useTrackProcessor(): ProcessorState { - const state = use(ProcessorContext); - if (state === undefined) - throw new Error( - "useTrackProcessor must be used within a ProcessorProvider", - ); - return state; + return useBehavior(useTrackProcessorState$()); } export function useTrackProcessorState$(): Behavior { - return useValueBehavior(useTrackProcessor()); + const effects = use(ProcessorContext); + if (effects === undefined) + throw new Error( + "useTrackProcessor must be used within a ProcessorProvider", + ); + return effects.state$; } /** @@ -104,8 +106,13 @@ export function applyProcessor( logger.debug("Not attaching video processor to an ended track"); return; } + const track = videoTrack.mediaStreamTrack; videoTrack.setProcessor(convertToLivekitProcessor(processor)).catch((e) => { - logger.warn("Failed to attach video processor", e); + // The pipeline builds one track at a time, so a track can end while its + // build waits a turn. The camera's next track attaches it again. + if (track.readyState === "ended") + logger.debug("Video processor not attached: the track ended first"); + else logger.warn("Failed to attach video processor", e); }); } if (!processor && videoTrack.getProcessor()) { @@ -151,28 +158,22 @@ function supportsBackgroundProcessors(): boolean { } export const ProcessorProvider: FC = ({ children }) => { - // The setting the user wants to have - const [blurActivated] = useSetting(backgroundBlurSettings); - const supported = useMemo(() => supportsBackgroundProcessors(), []); - const blur = useMemo( - () => - fromLivekitProcessor( - new ProcessorWrapper( - new BlurBackgroundTransformer({ blurRadius: 15 }), - "background-blur", + const [effects, setEffects] = useState(null); + useEffect(() => { + const scope = new ObservableScope(); + setEffects( + new BackgroundEffects(scope, { + supported: supportsBackgroundProcessors(), + blur$: backgroundBlurSettings.value$, + pipeline: new OneStepPipeline( + new BackgroundEffectTransformer({ backgroundDisabled: true }), + "background-effect", ), - ), - [], - ); + }), + ); + return (): void => scope.end(); + }, []); - // This is the actual state exposed through the context - const processorState = useMemo( - () => ({ - supported, - processor: supported && blurActivated ? blur : undefined, - }), - [supported, blurActivated, blur], - ); - - return {children}; + if (effects === null) return null; + return {children}; };