diff --git a/packages/core/README.md b/packages/core/README.md index 8576312d..81a569d2 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -182,7 +182,7 @@ const response: StandardResponse = { } ``` -Events are interpreted as follows: `yield` emits a `message`, `throw` emits an `error`, and `return` emits a `close` event. Note that `close` does not cause [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) to close the connection because it is not part of the SSE specification. However, when using Standard Server for client-side streaming, `close` is treated as the end of the stream, so the connection is closed and no reconnection is attempted. +Events are interpreted as follows: `yield` emits a `message`, `throw` emits an `error`, and `return` emits a `close` event. Note that `close` does not cause [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) to close the connection because it is not part of the SSE specification. However, when using Standard Server for client-side streaming, `close` is treated as the end of the stream, so the connection is closed and no reconnection is attempted. A stream that ends without a `close` or `error` event is treated as cut off, so reading it throws an `EventStreamDecoderError`. For explicit SSE encoding, decoding, and metadata handling, see the [Event-Stream Helpers](#event-stream-helpers) below. diff --git a/packages/fetch/src/event-stream.test.ts b/packages/fetch/src/event-stream.test.ts index 8602cbbc..ca0d7ca9 100644 --- a/packages/fetch/src/event-stream.test.ts +++ b/packages/fetch/src/event-stream.test.ts @@ -1,4 +1,9 @@ -import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core' +import { + ErrorEvent, + EventStreamDecoderError, + getEventMeta, + withEventMeta, +} from '@standard-server/core' import { isAsyncIteratorObject, sleep } from '@standard-server/shared' import { toAsyncIteratorObject, toEventStream } from './event-stream' @@ -82,17 +87,30 @@ describe('toAsyncIteratorObject', () => { return true }) - expect(await generator.next()).toSatisfy(({ done, value }) => { - expect(done).toEqual(true) - expect(value).toEqual(undefined) - expect(getEventMeta(value)).toEqual(undefined) - - return true - }) + // ending without a close event is treated like ending with an incomplete message + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) await expect(stream.getReader().closed).resolves.toBe(undefined) }) + it('with empty stream', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.close() + }, + }).pipeThrough(new TextEncoderStream()) + + const generator = toAsyncIteratorObject(stream) + + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) + }) + it('with extra blank lines', async () => { const stream = new ReadableStream({ async pull(controller) { @@ -100,6 +118,7 @@ describe('toAsyncIteratorObject', () => { controller.enqueue('event: message\ndata: {"order": 1}\n\n\n') controller.enqueue('\r\n: ping\r\n\r\n\r\n') controller.enqueue('event: message\ndata: {"order": 2}\n\n\n') + controller.enqueue('\n\nevent: close\n\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) @@ -120,6 +139,7 @@ describe('toAsyncIteratorObject', () => { controller.enqueue('event:\ndata: {"order": 3}\nid: id-3\n\n') controller.enqueue('data: 4\n\n') controller.enqueue('data:\n\n') + controller.enqueue('event: close\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) @@ -149,11 +169,13 @@ describe('toAsyncIteratorObject', () => { expect(await generator.next()).toEqual({ done: true, value: undefined }) }) - it('with empty stream', async () => { + it('with null stream', async () => { const generator = toAsyncIteratorObject(null) expect(generator).toSatisfy(isAsyncIteratorObject) - expect(await generator.next()).toEqual({ done: true }) - expect(await generator.next()).toEqual({ done: true }) + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) }) it.skipIf(process.version.startsWith('v26.'))('with error event', async () => { diff --git a/packages/fetch/src/event-stream.ts b/packages/fetch/src/event-stream.ts index 94462f88..4834147f 100644 --- a/packages/fetch/src/event-stream.ts +++ b/packages/fetch/src/event-stream.ts @@ -1,6 +1,7 @@ import { encodeEventStreamMessage, ErrorEvent, + EventStreamDecoderError, EventStreamDecoderStream, getEventMeta, unwrapEvent, @@ -25,21 +26,20 @@ export function toAsyncIteratorObject( return new AsyncIteratorClass( async () => { while (true) { - if (reader === undefined) { - return { done: true, value: undefined } - } - - const { done, value } = await reader.read() + const result = await reader?.read() /** - * The sender closed the stream without sending a 'close' event. + * A stream only completes with a 'close' or 'error' event, so ending without + * one means it was cut off, like a stream ending with an incomplete message. * A read cancelled by `return()` also ends here, and AsyncIteratorClass - * resolves it as done either way. + * resolves it as done instead. */ - if (done) { - return { done: true, value: undefined } + if (result === undefined || result.done) { + throw new EventStreamDecoderError('Event Stream ended before complete') } + const { value } = result + switch (value.event) { case 'message': { let message = parseEmptyableJSON(value.data) diff --git a/packages/node/src/event-stream.test.ts b/packages/node/src/event-stream.test.ts index ad61a29a..2a576c6a 100644 --- a/packages/node/src/event-stream.test.ts +++ b/packages/node/src/event-stream.test.ts @@ -20,6 +20,7 @@ it('toAsyncIteratorObject', async () => { controller.enqueue('event: message\ndata: 1\n\n') controller.enqueue('event: message\ndata: 2\n\n') controller.enqueue('event: message\ndata: 3\n\n') + controller.enqueue('event: close\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream())