Function collectSSEEvents

apps/sim/lib/workflows/streaming/streaming.test.ts:46–83

(
  stream: ReadableStream<Uint8Array>
)

}

async function collectSSEEvents(
  stream: ReadableStream<Uint8Array>
): Promise<Record<string, unknown>[]> {
  const reader = stream.getReader()
  const decoder = new TextDecoder()
  const events: Record<string, unknown>[] = []
  let buffer = ''

  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) {
        buffer += decoder.decode()
        break
      }
      buffer += decoder.decode(value, { stream: true })
    }
  } finally {
    reader.releaseLock()
  }

  for (const chunk of buffer.split('\n\n')) {
    if (!chunk.startsWith('data: ')) {
      continue
    }
    const payload = chunk.substring(6)
    if (payload === '[DONE]') {
      continue
    }
    const event = JSON.parse(payload) as unknown
    if (event === '[DONE]') {
      continue
    }
    events.push(event as Record<string, unknown>)
  }

  return events
}

describe('createStreamingResponse', () => {
  beforeEach(() => {