fix: let only the latest subscription detach a camera track

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
 

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();
+  });
 });

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<CameraFrameRuntime> | null = null;
 const activeSubscriptions = new WeakMap<CameraFrameProcessor, CameraFrameSubscription>();
+// A track has one native consumer slot, shared by all its handles, so only its latest subscription may detach.
+const latestSubscriptionByTrack = new Map<string, CameraFrameSubscription>();
 
 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,13 +166,17 @@ 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();
       });
     },
   };
   activeSubscriptions.set(processor, subscription);
+  latestSubscriptionByTrack.set(processor.trackId, subscription);
   return subscription;
 }