diff --git a/.changeset/wild-dingos-doubt.md b/.changeset/wild-dingos-doubt.md new file mode 100644 index 000000000..d56f6a224 --- /dev/null +++ b/.changeset/wild-dingos-doubt.md @@ -0,0 +1,5 @@ +--- +"@whereby.com/assistant-sdk": minor +--- + +Gracefully shutdown AudioMixer diff --git a/packages/assistant-sdk/src/AudioMixer/__tests__/AudioMixer.spec.ts b/packages/assistant-sdk/src/AudioMixer/__tests__/AudioMixer.spec.ts index f5e75f00a..e8417dffe 100644 --- a/packages/assistant-sdk/src/AudioMixer/__tests__/AudioMixer.spec.ts +++ b/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"); diff --git a/packages/assistant-sdk/src/AudioMixer/index.ts b/packages/assistant-sdk/src/AudioMixer/index.ts index d41e47a3d..3c984bcf7 100644 --- a/packages/assistant-sdk/src/AudioMixer/index.ts +++ b/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; + } } diff --git a/packages/assistant-sdk/src/utils/AudioEndpoints.ts b/packages/assistant-sdk/src/utils/AudioEndpoints.ts index 3c3d32c60..f797bb352 100644 --- a/packages/assistant-sdk/src/utils/AudioEndpoints.ts +++ b/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; }; } } diff --git a/packages/assistant-sdk/src/utils/ffmpeg-helpers.ts b/packages/assistant-sdk/src/utils/ffmpeg-helpers.ts index 8c7e41a85..bd448eecb 100644 --- a/packages/assistant-sdk/src/utils/ffmpeg-helpers.ts +++ b/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,