From a282f9f84ec118dc994f77691d8cce0bef3ae3e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Szymon=20Kozio=C5=82?= Date: Thu, 5 Nov 2026 12:36:57 +0100 Subject: [PATCH] fix: stop stale remove() detaching new consumer A track has one native consumer slot, shared by all processor handles. react-client starts the new effect before it stops the old one, so remove() on the old subscription emptied the slot and froze the video. Detach only from the latest subscription of the track. --- README.md | 6 +- .../attachCameraFrameCallback.test.ts | 87 ++++++++++++++++++- src/index.ts | 17 +++- 3 files changed, 103 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index b340262..99281d6 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,11 @@ subscription.remove(); The callback runs on a dedicated camera frame runtime, never on the JS thread. Frames are admitted one at a time: while the callback holds a frame, newer ones are dropped. Each processor takes one callback at a time; remove it before attaching another to the same -processor. Different processors (different camera tracks) can each have their own. +processor. Different camera tracks can each have their own. `getCameraFrameProcessor` +returns a new processor on each call, and a track delivers frames only to the last +callback attached to it. So a new processor for the same track takes over, and `remove()` +on the older subscription leaves the newer one attached. If the newer subscription is +removed first, the older one does not get frames again: remove it and attach again. ## Troubleshooting diff --git a/src/__tests__/attachCameraFrameCallback.test.ts b/src/__tests__/attachCameraFrameCallback.test.ts index 79f258b..6cdcfca 100644 --- a/src/__tests__/attachCameraFrameCallback.test.ts +++ b/src/__tests__/attachCameraFrameCallback.test.ts @@ -33,8 +33,25 @@ function createConsumer(): MockConsumer { return { bindRuntime: jest.fn(), setCallback: jest.fn(), clearCallback: jest.fn() }; } -function createProcessor(): MockProcessor { - return { attach: jest.fn(), detach: jest.fn() } as unknown as MockProcessor; +function createProcessor(trackId = 'camera-track'): MockProcessor { + return { trackId, attach: jest.fn(), detach: jest.fn() } as unknown as MockProcessor; +} + +function createCameraTrack(trackId = 'camera-track') { + let attachedConsumer: unknown = null; + return { + attachedConsumer: () => attachedConsumer, + getProcessor(): MockProcessor { + const processor = createProcessor(trackId); + processor.attach.mockImplementation((consumer: unknown) => { + attachedConsumer = consumer; + }); + processor.detach.mockImplementation(() => { + attachedConsumer = null; + }); + return processor; + }, + }; } const frameCallback: Library.CameraFrameCallback = () => {}; @@ -144,7 +161,9 @@ describe('attachCameraFrameCallback', () => { await expect(harness.attachCameraFrameCallback(processor, frameCallback)).rejects.toThrow( 'A camera frame callback is already attached.', ); - await expect(harness.attachCameraFrameCallback(createProcessor(), frameCallback)).resolves.toBeDefined(); + await expect( + harness.attachCameraFrameCallback(createProcessor('other-track'), frameCallback), + ).resolves.toBeDefined(); }); test('rejects a concurrent attach on the same processor while install is pending', async () => { @@ -169,4 +188,66 @@ describe('attachCameraFrameCallback', () => { await expect(harness.attachCameraFrameCallback(processor, frameCallback)).resolves.toBeDefined(); expect(processor.attach).toHaveBeenCalledTimes(2); }); + + describe('when a new processor handle replaces a subscription on the same track', () => { + test('removing the replaced subscription leaves the newer consumer attached', async () => { + const harness = loadLibrary(); + const track = createCameraTrack(); + + const replaced = await harness.attachCameraFrameCallback(track.getProcessor(), frameCallback); + await harness.attachCameraFrameCallback(track.getProcessor(), frameCallback); + const [replacedConsumer, newerConsumer] = harness.createConsumer.mock.results.map((result) => result.value); + replaced.remove(); + + expect(track.attachedConsumer()).toBe(newerConsumer); + expect(replacedConsumer.clearCallback).toHaveBeenCalledTimes(1); + expect(newerConsumer.clearCallback).not.toHaveBeenCalled(); + }); + + test('removing the newer subscription detaches the track', async () => { + const harness = loadLibrary(); + const track = createCameraTrack(); + + const replaced = await harness.attachCameraFrameCallback(track.getProcessor(), frameCallback); + const newer = await harness.attachCameraFrameCallback(track.getProcessor(), frameCallback); + const newerConsumer = harness.createConsumer.mock.results[1].value; + replaced.remove(); + newer.remove(); + + expect(track.attachedConsumer()).toBeNull(); + expect(newerConsumer.clearCallback).toHaveBeenCalledTimes(1); + }); + + test('a failed attach leaves the earlier subscription able to detach', async () => { + const harness = loadLibrary(); + const track = createCameraTrack(); + const failingProcessor = track.getProcessor(); + failingProcessor.attach.mockImplementation(() => { + throw new Error('attach failed'); + }); + + const earlier = await harness.attachCameraFrameCallback(track.getProcessor(), frameCallback); + await expect(harness.attachCameraFrameCallback(failingProcessor, frameCallback)).rejects.toThrow('attach failed'); + earlier.remove(); + + expect(track.attachedConsumer()).toBeNull(); + }); + }); + + test('removing a subscription does not affect another track', async () => { + const harness = loadLibrary(); + const firstTrack = createCameraTrack('first-track'); + const secondTrack = createCameraTrack('second-track'); + + const first = await harness.attachCameraFrameCallback(firstTrack.getProcessor(), frameCallback); + const second = await harness.attachCameraFrameCallback(secondTrack.getProcessor(), frameCallback); + const secondConsumer = harness.createConsumer.mock.results[1].value; + first.remove(); + + expect(firstTrack.attachedConsumer()).toBeNull(); + expect(secondTrack.attachedConsumer()).toBe(secondConsumer); + + second.remove(); + expect(secondTrack.attachedConsumer()).toBeNull(); + }); }); diff --git a/src/index.ts b/src/index.ts index f67bc41..068e6be 100644 --- a/src/index.ts +++ b/src/index.ts @@ -43,7 +43,10 @@ export interface CameraFrame { export type CameraFrameCallback = (frame: CameraFrame) => void; export interface CameraFrameSubscription { - /** Detaches from the track and clears the callback. Safe to repeat. */ + /** + * Always clears the callback. Detaches from the track only if no newer + * subscription took it over. Safe to repeat. + */ remove(): void; } @@ -70,6 +73,8 @@ interface CameraFrameRuntime { let runtimePromise: Promise | null = null; const activeSubscriptions = new WeakMap(); +// A track has one native consumer slot, shared by all its handles, so only its latest subscription may detach. +const latestSubscriptionByTrack = new Map(); function nativeModule(): FishjamWorkletsNativeModule { const module = NativeFishjamWorklets; @@ -122,7 +127,9 @@ function throwIfAttached(processor: CameraFrameProcessor): void { * Starts calling `callback` on the camera frame runtime for every frame the * processor's track captures. Each processor can have one callback at a time; * remove the previous subscription before attaching another to the same - * processor. Different processors may each have their own. + * processor. Different camera tracks may each have their own. A track delivers + * frames only to the last callback attached to it, so a new processor for the + * same track takes over from the earlier one. */ export async function attachCameraFrameCallback( processor: CameraFrameProcessor, @@ -159,7 +166,10 @@ export async function attachCameraFrameCallback( if (activeSubscriptions.get(processor) === subscription) { activeSubscriptions.delete(processor); } - processor.detach(); + if (latestSubscriptionByTrack.get(processor.trackId) === subscription) { + latestSubscriptionByTrack.delete(processor.trackId); + processor.detach(); + } scheduleOnRuntime(runtime, () => { 'worklet'; consumer.clearCallback(); @@ -167,5 +177,6 @@ export async function attachCameraFrameCallback( }, }; activeSubscriptions.set(processor, subscription); + latestSubscriptionByTrack.set(processor.trackId, subscription); return subscription; }