feat(stage-ui): introduce helpers for stream and tts chunking (#335)

packages/stage-ui/src/composables/queues.ts

@@ -4,7 +4,8 @@ import type { UseQueueReturn } from './queue'
 import { sleep } from '@moeru/std'
 
 import { EMOTION_VALUES } from '../constants/emotions'
-import { chunkTTSInput } from '../utils/tts'
+import { createControllableStream } from '../utils/stream'
+import { chunkToTTSQueue } from '../utils/tts'
 import { useQueue } from './queue'
 
 export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>) {
@@ -102,28 +103,14 @@ export function useDelayMessageQueue() {
 
 export function useMessageContentQueue(ttsQueue: UseQueueReturn<string>) {
   const encoder = new TextEncoder()
-  let enqueue: (data: Uint8Array) => void
-  const stream = new ReadableStream<Uint8Array>({
-    start(controller) {
-      enqueue = data => controller.enqueue(data)
-    },
-  });
+  const { stream, controller } = createControllableStream<Uint8Array>()
 
-  (async () => {
-    try {
-      for await (const chunk of chunkTTSInput(stream.getReader())) {
-        await ttsQueue.add(chunk.text)
-      }
-    }
-    catch (e) {
-      console.error('Error chunking input stream for TTS:', e)
-    }
-  })()
+  chunkToTTSQueue(stream.getReader(), ttsQueue)
 
   return useQueue<string>({
     handlers: [
       async (ctx) => {
-        enqueue(encoder.encode(ctx.data))
+        controller.enqueue(encoder.encode(ctx.data))
       },
     ],
   })

packages/stage-ui/src/utils/stream.ts

@@ -0,0 +1,16 @@
+export interface ControllableStream<R = any> {
+  stream: ReadableStream<R>
+  controller: ReadableStreamDefaultController<R>
+}
+
+export function createControllableStream<R = any>(): ControllableStream<R> {
+  // WHY!: ReadableStream.start is called synchronously and immediately
+  let controller!: ReadableStreamDefaultController<R>
+
+  const stream = new ReadableStream<R>({
+    start(ctrl) {
+      controller = ctrl
+    },
+  })
+  return { stream, controller }
+}

packages/stage-ui/src/utils/tts.ts

@@ -1,5 +1,7 @@
 import type { ReaderLike } from 'clustr'
 
+import type { UseQueueReturn } from '../composables/queue'
+
 import { readGraphemeClusters } from 'clustr'
 
 // A special character to instruct the TTS pipeline to flush
@@ -144,3 +146,14 @@ export async function* chunkTTSInput(input: string | ReaderLike, options?: TTSIn
     }
   }
 }
+
+export async function chunkToTTSQueue(reader: ReaderLike, queue: UseQueueReturn<string>) {
+  try {
+    for await (const chunk of chunkTTSInput(reader)) {
+      await queue.add(chunk.text)
+    }
+  }
+  catch (e) {
+    console.error('Error chunking stream to TTS queue:', e)
+  }
+}