feat(openai): implement close() for openai TTS (#883)

Author: KrishnaShuk

.changeset/vast-colts-burn.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents-plugin-openai': patch
+---
+
+Implemented close() in openai TTS

plugins/openai/src/tts.ts

@@ -30,6 +30,7 @@ export class TTS extends tts.TTS {
   #opts: TTSOptions;
   #client: OpenAI;
   label = 'openai.TTS';
+  private abortController = new AbortController();
 
   /**
    * Create a new instance of OpenAI TTS.
@@ -62,20 +63,27 @@ export class TTS extends tts.TTS {
     return new ChunkedStream(
       this,
       text,
-      this.#client.audio.speech.create({
-        input: text,
-        model: this.#opts.model,
-        voice: this.#opts.voice,
-        instructions: this.#opts.instructions,
-        response_format: 'pcm',
-        speed: this.#opts.speed,
-      }),
+      this.#client.audio.speech.create(
+        {
+          input: text,
+          model: this.#opts.model,
+          voice: this.#opts.voice,
+          instructions: this.#opts.instructions,
+          response_format: 'pcm',
+          speed: this.#opts.speed,
+        },
+        { signal: this.abortController.signal },
+      ),
     );
   }
 
   stream(): tts.SynthesizeStream {
     throw new Error('Streaming is not supported on OpenAI TTS');
   }
+
+  async close(): Promise<void> {
+    this.abortController.abort();
+  }
 }
 
 export class ChunkedStream extends tts.ChunkedStream {
@@ -89,25 +97,34 @@ export class ChunkedStream extends tts.ChunkedStream {
   }
 
   protected async run() {
-    const buffer = await this.stream.then((r) => r.arrayBuffer());
-    const requestId = shortuuid();
-    const audioByteStream = new AudioByteStream(OPENAI_TTS_SAMPLE_RATE, OPENAI_TTS_CHANNELS);
-    const frames = audioByteStream.write(buffer);
-
-    let lastFrame: AudioFrame | undefined;
-    const sendLastFrame = (segmentId: string, final: boolean) => {
-      if (lastFrame) {
-        this.queue.put({ requestId, segmentId, frame: lastFrame, final });
-        lastFrame = undefined;
+    try {
+      const buffer = await this.stream.then((r) => r.arrayBuffer());
+      const requestId = shortuuid();
+      const audioByteStream = new AudioByteStream(OPENAI_TTS_SAMPLE_RATE, OPENAI_TTS_CHANNELS);
+      const frames = audioByteStream.write(buffer);
+
+      let lastFrame: AudioFrame | undefined;
+      const sendLastFrame = (segmentId: string, final: boolean) => {
+        if (lastFrame) {
+          this.queue.put({ requestId, segmentId, frame: lastFrame, final });
+          lastFrame = undefined;
+        }
+      };
+
+      for (const frame of frames) {
+        sendLastFrame(requestId, false);
+        lastFrame = frame;
       }
-    };
+      sendLastFrame(requestId, true);
 
-    for (const frame of frames) {
-      sendLastFrame(requestId, false);
-      lastFrame = frame;
+      this.queue.close();
+    } catch (error) {
+      if (error instanceof Error && error.name === 'AbortError') {
+        return;
+      }
+      throw error;
+    } finally {
+      this.queue.close();
     }
-    sendLastFrame(requestId, true);
-
-    this.queue.close();
   }
 }