assistant-sdk: stop active participant slots with mixer

.changeset/wild-dingos-doubt.md

@@ -0,0 +1,5 @@
+---
+"@whereby.com/assistant-sdk": minor
+---
+
+Gracefully shutdown AudioMixer

packages/assistant-sdk/src/AudioMixer/__tests__/AudioMixer.spec.ts

@@ -23,6 +23,14 @@ const mixerInstance = {
     clearSlotQueue: jest.fn(),
 };
 
+const mockCombinedAudioTrack = {
+    id: "mock-track-id",
+    kind: "audio",
+    addEventListener: jest.fn(),
+    removeEventListener: jest.fn(),
+    stop: jest.fn(),
+};
+
 jest.mock("../../utils/ffmpeg-helpers", () => ({
     createFfmpegMixer: jest.fn(() => mixerInstance),
     MIXER_SLOTS: 20,
@@ -36,12 +44,7 @@ jest.mock("@roamhq/wrtc", () => ({
     })),
     nonstandard: {
         RTCAudioSource: jest.fn().mockImplementation(() => ({
-            createTrack: jest.fn().mockReturnValue({
-                id: "mock-track-id",
-                kind: "audio",
-                addEventListener: jest.fn(),
-                removeEventListener: jest.fn(),
-            }),
+            createTrack: jest.fn().mockReturnValue(mockCombinedAudioTrack),
         })),
     },
 }));
@@ -646,6 +649,26 @@ describe("AudioMixer", () => {
             expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
         });
 
+        it("should stop all active slots", () => {
+            const participant1 = createMockParticipant({ id: "p1" });
+            const participant2 = createMockParticipant({ id: "p2" });
+            const screenshare1 = createMockScreenshare({ id: "s1" });
+            const screenshare2 = createMockScreenshare({ id: "s2" });
+            audioMixer.handleRemoteParticipants([participant1, participant2]);
+            audioMixer.handleScreenshares([screenshare1, screenshare2]);
+
+            audioMixer.stopAudioMixer();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+            expect(mockSlotBinding.stop).toHaveBeenCalledTimes(4);
+        });
+
+        it("should stop the combined audio stream tracks", () => {
+            audioMixer.stopAudioMixer();
+
+            expect(mockCombinedAudioTrack.stop).toHaveBeenCalled();
+        });
+
         it("should clear all participant slots", () => {
             const participants = [createMockParticipant({ id: "p1" }), createMockParticipant({ id: "p2" })];
             audioMixer.handleRemoteParticipants(participants);
@@ -676,6 +699,186 @@ describe("AudioMixer", () => {
         });
     });
 
+    describe("destroy", () => {
+        it("should stop FFmpeg process if running", () => {
+            const participant = createMockParticipant({ id: "p1" });
+            audioMixer.handleRemoteParticipants([participant]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+        });
+
+        it("should stop all active slots", () => {
+            const participant1 = createMockParticipant({ id: "p1" });
+            const participant2 = createMockParticipant({ id: "p2" });
+            const screenshare1 = createMockScreenshare({ id: "s1" });
+            const screenshare2 = createMockScreenshare({ id: "s2" });
+            audioMixer.handleRemoteParticipants([participant1, participant2]);
+            audioMixer.handleScreenshares([screenshare1, screenshare2]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+            expect(mockSlotBinding.stop).toHaveBeenCalledTimes(4);
+        });
+
+        it("should stop the combined audio stream tracks", () => {
+            audioMixer.destroy();
+
+            expect(mockCombinedAudioTrack.stop).toHaveBeenCalled();
+        });
+
+        it("should clear all participant slots", () => {
+            const participants = [createMockParticipant({ id: "p1" }), createMockParticipant({ id: "p2" })];
+            audioMixer.handleRemoteParticipants(participants);
+
+            audioMixer.destroy();
+
+            const newParticipant = createMockParticipant({ id: "p3" });
+            audioMixer.handleRemoteParticipants([newParticipant]);
+
+            expect(mixerInstance.writeAudioDataToFFmpeg).toHaveBeenLastCalledWith(
+                expect.any(Object),
+                0,
+                expect.any(Object),
+            );
+        });
+
+        it("should not recreate media stream", () => {
+            const streamBefore = audioMixer.getCombinedAudioStream();
+            audioMixer.destroy();
+            const streamAfter = audioMixer.getCombinedAudioStream();
+
+            expect(streamBefore).toBeDefined();
+            expect(streamAfter).toEqual(null);
+        });
+
+        it("should handle being called when no FFmpeg process is running", () => {
+            expect(() => audioMixer.destroy()).not.toThrow();
+        });
+    });
+
+    describe("destroy", () => {
+        it("should stop FFmpeg process if running", () => {
+            const participant = createMockParticipant({ id: "p1" });
+            audioMixer.handleRemoteParticipants([participant]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+        });
+
+        it("should stop all active slots", () => {
+            const participant1 = createMockParticipant({ id: "p1" });
+            const participant2 = createMockParticipant({ id: "p2" });
+            const screenshare1 = createMockScreenshare({ id: "s1" });
+            const screenshare2 = createMockScreenshare({ id: "s2" });
+            audioMixer.handleRemoteParticipants([participant1, participant2]);
+            audioMixer.handleScreenshares([screenshare1, screenshare2]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+            expect(mockSlotBinding.stop).toHaveBeenCalledTimes(4);
+        });
+
+        it("should stop the combined audio stream tracks", () => {
+            audioMixer.destroy();
+
+            expect(mockCombinedAudioTrack.stop).toHaveBeenCalled();
+        });
+
+        it("should clear all participant slots", () => {
+            const participants = [createMockParticipant({ id: "p1" }), createMockParticipant({ id: "p2" })];
+            audioMixer.handleRemoteParticipants(participants);
+
+            audioMixer.destroy();
+
+            const newParticipant = createMockParticipant({ id: "p3" });
+            audioMixer.handleRemoteParticipants([newParticipant]);
+
+            expect(mixerInstance.writeAudioDataToFFmpeg).toHaveBeenLastCalledWith(
+                expect.any(Object),
+                0,
+                expect.any(Object),
+            );
+        });
+
+        it("should not recreate media stream", () => {
+            const streamBefore = audioMixer.getCombinedAudioStream();
+            audioMixer.destroy();
+            const streamAfter = audioMixer.getCombinedAudioStream();
+
+            expect(streamBefore).toBeDefined();
+            expect(streamAfter).toEqual(null);
+        });
+
+        it("should handle being called when no FFmpeg process is running", () => {
+            expect(() => audioMixer.destroy()).not.toThrow();
+        });
+    });
+
+    describe("destroy", () => {
+        it("should stop FFmpeg process if running", () => {
+            const participant = createMockParticipant({ id: "p1" });
+            audioMixer.handleRemoteParticipants([participant]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+        });
+
+        it("should stop all active slots", () => {
+            const participant1 = createMockParticipant({ id: "p1" });
+            const participant2 = createMockParticipant({ id: "p2" });
+            const screenshare1 = createMockScreenshare({ id: "s1" });
+            const screenshare2 = createMockScreenshare({ id: "s2" });
+            audioMixer.handleRemoteParticipants([participant1, participant2]);
+            audioMixer.handleScreenshares([screenshare1, screenshare2]);
+
+            audioMixer.destroy();
+
+            expect(mixerInstance.stopFFmpegProcess).toHaveBeenCalledWith(mockFFmpegProcess);
+            expect(mockSlotBinding.stop).toHaveBeenCalledTimes(4);
+        });
+
+        it("should stop the combined audio stream tracks", () => {
+            audioMixer.destroy();
+
+            expect(mockCombinedAudioTrack.stop).toHaveBeenCalled();
+        });
+
+        it("should clear all participant slots", () => {
+            const participants = [createMockParticipant({ id: "p1" }), createMockParticipant({ id: "p2" })];
+            audioMixer.handleRemoteParticipants(participants);
+
+            audioMixer.destroy();
+
+            const newParticipant = createMockParticipant({ id: "p3" });
+            audioMixer.handleRemoteParticipants([newParticipant]);
+
+            expect(mixerInstance.writeAudioDataToFFmpeg).toHaveBeenLastCalledWith(
+                expect.any(Object),
+                0,
+                expect.any(Object),
+            );
+        });
+
+        it("should not recreate media stream", () => {
+            const streamBefore = audioMixer.getCombinedAudioStream();
+            audioMixer.destroy();
+            const streamAfter = audioMixer.getCombinedAudioStream();
+
+            expect(streamBefore).toBeDefined();
+            expect(streamAfter).toEqual(null);
+        });
+
+        it("should handle being called when no FFmpeg process is running", () => {
+            expect(() => audioMixer.destroy()).not.toThrow();
+        });
+    });
+
     describe("error handling", () => {
         it("should handle writeAudioDataToFFmpeg throwing", () => {
             const error = new Error("FFmpeg write failed");

packages/assistant-sdk/src/AudioMixer/index.ts

@@ -125,10 +125,9 @@ export class AudioMixer extends EventEmitter {
 
     public handleRemoteParticipants(participants: RemoteParticipantState[]): void {
         const liveIds = new Set(participants.map((p) => p.id).filter(Boolean) as string[]);
-        const typedSlots = this.slotsByType("participant");
 
         // eslint-disable-next-line @typescript-eslint/no-unused-vars
-        for (const [slot, pid] of typedSlots) {
+        for (const [slot, pid] of this.slotsByType("participant")) {
             if (pid && !liveIds.has(pid)) this.detachMixable(pid);
         }
 
@@ -150,7 +149,7 @@ export class AudioMixer extends EventEmitter {
         );
         const liveIds = new Set(screensharesWithAudio.map((p) => p.id).filter(Boolean) as string[]);
         // eslint-disable-next-line @typescript-eslint/no-unused-vars
-        for (const [slot, sid] of this.slotsByType("screenshare")) {
+        for (const [, sid] of this.slotsByType("screenshare")) {
             if (sid && !liveIds.has(sid)) this.detachMixable(sid);
         }
 
@@ -177,9 +176,19 @@ export class AudioMixer extends EventEmitter {
             this.mixer.stopFFmpegProcess(this.ffmpegProcess);
             this.ffmpegProcess = null;
         }
+
+        Object.values(this.activeSlots).forEach((slot) => slot?.stop());
+        this.combinedAudioStream?.getTracks().forEach((track) => track.stop());
         this.mixableSlots = new Map(Array.from({ length: MIXER_SLOTS }, (_, i) => [i, ""]));
         this.activeSlots = {};
         // Recreate the media stream to avoid stale references
         this.setupMediaStream();
     }
+
+    public destroy(): void {
+        this.stopAudioMixer();
+
+        this.combinedAudioStream = null;
+        this.rtcAudioSource = null;
+    }
 }

packages/assistant-sdk/src/utils/AudioEndpoints.ts

@@ -7,13 +7,6 @@ const {
 export class AudioSource extends RTCAudioSource {}
 
 export class AudioSink extends RTCAudioSink {
-    private _sink: wrtc.nonstandard.RTCAudioSink;
-
-    constructor(track: MediaStreamTrack) {
-        super(track);
-        this._sink = new RTCAudioSink(track);
-    }
-
     subscribe(
         cb: (d: {
             samples: Int16Array;
@@ -23,9 +16,9 @@ export class AudioSink extends RTCAudioSink {
             numberOfFrames?: number;
         }) => void,
     ) {
-        this._sink.ondata = cb;
+        this.ondata = cb;
         return () => {
-            this._sink.ondata = undefined;
+            this.ondata = undefined;
         };
     }
 }

packages/assistant-sdk/src/utils/ffmpeg-helpers.ts

@@ -444,27 +444,54 @@ export function createFfmpegMixer() {
      */
     function stopFFmpegProcess(ffmpegProcess: ChildProcessWithoutNullStreams) {
         stopPacer();
-        if (ffmpegProcess && !ffmpegProcess.killed) {
+        if (!ffmpegProcess || ffmpegProcess.exitCode !== null) {
+            return;
+        }
+
+        try {
+            ffmpegProcess.stdout.unpipe();
+        } catch {
+            console.error("Failed to unpipe ffmpeg stdout");
+        }
+        for (let i = 0; i < MIXER_SLOTS; i++) {
+            const w = ffmpegProcess.stdio[3 + i] as Stream.Writable;
             try {
-                ffmpegProcess.stdout.unpipe();
+                w.end();
+                w.destroy();
             } catch {
-                console.error("Failed to unpipe ffmpeg stdout");
+                console.error("Failed to end ffmpeg writable stream");
             }
-            for (let i = 0; i < MIXER_SLOTS; i++) {
-                const w = ffmpegProcess.stdio[3 + i] as Stream.Writable;
+        }
+        try {
+            ffmpegProcess.stdout?.destroy?.();
+        } catch {
+            // noop as we are tearing down
+        }
+
+        try {
+            ffmpegProcess.stderr?.destroy?.();
+        } catch {
+            // noop as we are tearing down
+        }
+
+        try {
+            ffmpegProcess.stdin?.end();
+        } catch {
+            console.error("Failed to end ffmpeg stdin");
+        }
+        ffmpegProcess.kill("SIGTERM");
+
+        const t = setTimeout(() => {
+            if (ffmpegProcess.exitCode == null) {
                 try {
-                    w.end();
+                    ffmpegProcess.kill("SIGKILL");
                 } catch {
-                    console.error("Failed to end ffmpeg writable stream");
+                    // noop since we are tearing down
                 }
             }
-            try {
-                ffmpegProcess.stdin?.write("q\n");
-                ffmpegProcess.stdin?.end();
-            } catch {
-                console.error("Failed to end ffmpeg stdin");
-            }
-        }
+        }, 2000);
+
+        ffmpegProcess.once("exit", () => clearTimeout(t));
     }
     return {
         spawnFFmpegProcess,