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; }