diff --git a/.changeset/audio-mixer-end-input-wakeup.md b/.changeset/audio-mixer-end-input-wakeup.md new file mode 100644 index 00000000..caaa9d12 --- /dev/null +++ b/.changeset/audio-mixer-end-input-wakeup.md @@ -0,0 +1,5 @@ +--- +'@livekit/rtc-node': patch +--- + +`AudioMixer.endInput()` now ends iteration when no streams are left. Before, a consumer already waiting for a frame stayed blocked forever if `endInput()` was called after the last stream was removed (or before any was added). diff --git a/packages/livekit-rtc/src/audio_mixer.test.ts b/packages/livekit-rtc/src/audio_mixer.test.ts index d4dfbfaa..e987957f 100644 --- a/packages/livekit-rtc/src/audio_mixer.test.ts +++ b/packages/livekit-rtc/src/audio_mixer.test.ts @@ -111,6 +111,69 @@ describe('AudioMixer', () => { expect(frames.length).toBe(0); }); + it('ends iteration when endInput() is called while a consumer is waiting', async () => { + const mixer = new AudioMixer(48000, 1); + + // Start consuming before endInput(), so the consumer is already parked + // waiting for a frame when the mixer is told no more streams are coming. + const consumed = (async () => { + const frames: AudioFrame[] = []; + for await (const frame of mixer) { + frames.push(frame); + } + return frames; + })(); + + await new Promise((resolve) => setTimeout(resolve, 20)); + mixer.endInput(); + + const result = await Promise.race([ + consumed, + new Promise<'hung'>((resolve) => setTimeout(() => resolve('hung'), 1000)), + ]); + expect(result).toEqual([]); + + await mixer.aclose(); + }); + + it('ends iteration when endInput() follows removal of the last stream', async () => { + const mixer = new AudioMixer(48000, 1, { streamTimeoutMs: 10 }); + + // A stream that never produces audio, like a muted participant. + const silentStream = async function* (): AsyncGenerator { + await new Promise(() => {}); + }; + const stream = silentStream(); + + const originalWarn = console.warn; + console.warn = () => {}; + try { + mixer.addStream(stream); + + const consumed = (async () => { + const frames: AudioFrame[] = []; + for await (const frame of mixer) { + frames.push(frame); + } + return frames; + })(); + + await new Promise((resolve) => setTimeout(resolve, 30)); + mixer.removeStream(stream); + await new Promise((resolve) => setTimeout(resolve, 30)); + mixer.endInput(); + + const result = await Promise.race([ + consumed, + new Promise<'hung'>((resolve) => setTimeout(() => resolve('hung'), 1000)), + ]); + expect(result).toEqual([]); + } finally { + console.warn = originalWarn; + await mixer.aclose(); + } + }); + it('clips audio values to int16 range', async () => { const sampleRate = 48000; const numChannels = 1; diff --git a/packages/livekit-rtc/src/audio_mixer.ts b/packages/livekit-rtc/src/audio_mixer.ts index e2aaef84..248ba1e5 100644 --- a/packages/livekit-rtc/src/audio_mixer.ts +++ b/packages/livekit-rtc/src/audio_mixer.ts @@ -185,6 +185,10 @@ export class AudioMixer { */ endInput(): void { this.ending = true; + // No more streams will be added, so close the signal queue. If the mixer is + // parked waiting for a stream (none left), this wakes it so it can see it is + // ending, exit, and close the output queue for any waiting consumer. + this.streamSignal.close(); } private async getNextFrame(): Promise {