Function readSSEStream

apps/sim/lib/core/utils/sse.ts:216–271

(
  body: ReadableStream<Uint8Array>,
  options: ReadSSEStreamOptions = {}
)

 * @returns The accumulated content from the stream
 */
export async function readSSEStream(
  body: ReadableStream<Uint8Array>,
  options: ReadSSEStreamOptions = {}
): Promise<string> {
  const { onChunk, onAccumulated, signal } = options
  const reader = body.getReader()
  const decoder = new TextDecoder()
  let accumulatedContent = ''
  let buffer = ''

  try {
    while (true) {
      if (signal?.aborted) {
        break
      }

      const { done, value } = await reader.read()

      if (done) {
        const remaining = decoder.decode()
        if (remaining) {
          buffer += remaining
        }
        break
      }

      buffer += decoder.decode(value, { stream: true })
      const lines = buffer.split('\n\n')
      buffer = lines.pop() || ''

      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const lineData = line.substring(6)
          if (lineData === '[DONE]') continue

          try {
            const data = JSON.parse(lineData)
            if (data.error) throw new Error(data.error)
            if (data.chunk) {
              accumulatedContent += data.chunk
              onChunk?.(data.chunk)
              onAccumulated?.(accumulatedContent)
            }
            if (data.done) break
          } catch {
            // Skip unparseable lines
          }
        }
      }
    }
  } finally {
    reader.releaseLock()
  }

  return accumulatedContent
}