Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions cli/src/call/sse.test.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import { execFile } from 'node:child_process'
import { promisify } from 'node:util'
import { describe, expect, it } from 'vite-plus/test'
import { parseSSE } from './sse'

Expand Down Expand Up @@ -66,6 +68,46 @@ describe('parseSSE', () => {
expect((await collect(s))[0]?.data).toBe('😀')
})

it('propagates a response body error to the consumer', async () => {
const error = new Error('network disconnected')
const body = new ReadableStream<Uint8Array>({
start(controller) {
controller.error(error)
},
})
await expect(collect(body)).rejects.toBe(error)
})

it('observes read failures while the consumer is processing a yielded event', async () => {
// Isolate Node's strict unhandled-rejection policy from the test runner.
const source = new URL('./sse.ts', import.meta.url).href
const script = `
import assert from 'node:assert/strict'
import { setImmediate } from 'node:timers/promises'
import { parseSSE } from ${JSON.stringify(source)}
let controller
const body = new ReadableStream({ start(c) { controller = c } })
const events = parseSSE(body)
controller.enqueue(new TextEncoder().encode('data: first\\n\\ndata: second\\n\\n'))
const decoder = new TextDecoder()
assert.equal(decoder.decode((await events.next()).value.data), 'first')
const error = new Error('network disconnected')
controller.error(error)
await setImmediate()
assert.equal(decoder.decode((await events.next()).value.data), 'second')
await assert.rejects(events.next(), (failure) => failure === error)
console.log('stream error reached the consumer')
`
const { stdout, stderr } = await promisify(execFile)(process.execPath, [
'--unhandled-rejections=strict',
'--input-type=module',
'-e',
script,
])
expect(stdout.trim()).toBe('stream error reached the consumer')
expect(stderr).toBe('')
})

it('aborts when signal fires', async () => {
const ctrl = new AbortController()
const slow = new ReadableStream<Uint8Array>({
Expand Down
4 changes: 4 additions & 0 deletions cli/src/call/sse.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,10 @@ export async function* parseSSE(
}
})()

// Observe background failures even while a yielded event suspends the consumer.
// Awaiting the original pump below still propagates the error after the queue drains.
void pump.catch(() => {})

try {
while (true) {
while (queue.length > 0) {
Expand Down
Loading