diff --git a/packages/core/src/event-stream/decoder.test.ts b/packages/core/src/event-stream/decoder.test.ts index 46dc92b..4f01de7 100644 --- a/packages/core/src/event-stream/decoder.test.ts +++ b/packages/core/src/event-stream/decoder.test.ts @@ -143,7 +143,7 @@ describe('eventStreamDecoder', () => { }) it('emits the same events regardless of chunk size', () => { - const stream = 'event: a\r\ndata: 1\r\n\r\n: comment\ndata: 2\ndata: 3\n\nid: 9\rretry: 50\rdata: 4\r\revent: done\ndata: bye\n\n' + const stream = '\n\nevent: a\r\ndata: 1\r\n\r\n\r\n: comment\ndata: 2\ndata: 3\n\n\n\nid: 9\rretry: 50\rdata: 4\r\r\revent: done\ndata: bye\n\n\r\n' const expected = feedAll([stream]) expect(expected).toHaveLength(4) @@ -158,6 +158,20 @@ describe('eventStreamDecoder', () => { } }) + it('emits nothing for a stream of only blank lines', () => { + expect(feedAll(['\n\n\n'])).toEqual([]) + expect(feedAll(['\r', '\n', '\r\r\n', '', '\n'])).toEqual([]) + }) + + // Unlike the spec, comment-only messages are emitted (e.g. keep-alives), so + // skipping blank lines must not skip them. + it('emits comment-only messages surrounded by extra blank lines', () => { + expect(feedAll(['\n\n: ping\n\n\n', ': ', 'pong\r\n\r\n\r\n'])).toEqual([ + { comments: ['ping'] }, + { comments: ['pong'] }, + ]) + }) + it('decodes a large message fed in many small chunks', () => { const value = 'x'.repeat(64 * 1024) const stream = `event: big\ndata: ${value}\ndata: ${value}\n\n` @@ -174,9 +188,11 @@ describe('eventStreamDecoder', () => { }) describe('delimiters across chunk boundaries', () => { + // Per spec, a blank line with nothing buffered dispatches nothing, so + // leading and extra blank lines are ignored. it('handles every delimiter split at every position', () => { - for (const delimiter of ['\n\n', '\r\r', '\n\r', '\n\r\n', '\r\n\n', '\r\n\r\n']) { - const stream = `data: first${delimiter}data: second${delimiter}` + for (const delimiter of ['\n\n', '\r\r', '\n\r', '\n\r\n', '\r\n\n', '\r\n\r\n', '\n\n\n', '\r\r\r', '\r\n\r\n\r\n', '\n\r\n\r\n']) { + const stream = `${delimiter}data: first${delimiter}data: second${delimiter}` for (let split = 1; split < stream.length; split++) { const events = feedAll([stream.slice(0, split), stream.slice(split)]) @@ -218,7 +234,7 @@ describe('eventStreamDecoder', () => { ]) }) - it('keeps the CRLF discard window open across empty chunks', () => { + it('ignores blank lines after a delimiter across an empty chunk', () => { const events = feedAll([ 'data: first\n\r', '', @@ -268,6 +284,20 @@ describe('eventStreamDecoder', () => { { event: 'message', data: 'hello1\nworld' }, ]) }) + + it('throws when extra blank lines are followed by an incomplete message', () => { + const events: EventStreamMessage[] = [] + const decoder = new EventStreamDecoder(event => events.push(event)) + + decoder.feed('data: hello\n\n\n\r') + decoder.feed('\n\rdata: incomplete\n') + + expect(() => decoder.end()).toThrowError('Event Stream ended before complete') + + expect(events).toEqual([ + { data: 'hello' }, + ]) + }) }) }) diff --git a/packages/core/src/event-stream/decoder.ts b/packages/core/src/event-stream/decoder.ts index 9f255cf..2aaffff 100644 --- a/packages/core/src/event-stream/decoder.ts +++ b/packages/core/src/event-stream/decoder.ts @@ -3,15 +3,16 @@ import { EventStreamDecoderError } from './error' // A line ending is CR, LF or CRLF. const LINE_ENDING_REGEX = /\r\n|\r(?!\n)|\n/ -const MESSAGE_DELIMITER_REGEX = /(?:\r\n|\r(?!\n)|\n){2}/ -const MESSAGE_DELIMITER_GLOBAL_REGEX = /(?:\r\n|\r(?!\n)|\n){2}/g +// A message ends at a blank line; any extra blank lines after it are part of +// the same delimiter, since the spec treats them as no-ops. +const MESSAGE_DELIMITER_REGEX = /(?:\r\n|\r(?!\n)|\n){2,}/g +const LEADING_LINE_ENDINGS_REGEX = /^[\r\n]+/ -// A delimiter is at most 4 characters ('\r\n\r\n'), so one crossing a chunk -// boundary must start within the last 3 characters of what came before. -const MAX_DELIMITER_OVERLAP = 3 +// Pending text never contains a blank line, so it ends in at most one line +// ending ('\r\n'). A delimiter crossing a chunk boundary therefore starts +// within its last 2 characters. +const MAX_DELIMITER_OVERLAP = 2 -const CR = 0x0D -const LF = 0x0A const SPACE = 0x20 export function decodeEventStreamMessage(encoded: string): EventStreamMessage { @@ -68,13 +69,12 @@ export function decodeEventStreamMessage(encoded: string): EventStreamMessage { } export class EventStreamDecoder { + // The incomplete message: empty, or text that neither starts with a line + // ending nor contains a blank line. private pending: string[] = [] - // Last up-to-3 characters of the pending buffer, prefixed to the next chunk - // so a delimiter straddling the boundary is still found. + // Last MAX_DELIMITER_OVERLAP characters of the pending text, prefixed to the + // next chunk so a delimiter straddling the boundary is still found. private tail: string = '' - // Set when a chunk-ending '\r' was already consumed as a line ending, so a - // leading '\n' in the next chunk is the second half of that CRLF pair. - private discardLeadingLF: boolean = false constructor( private readonly onEvent: (event: EventStreamMessage) => void, @@ -82,55 +82,43 @@ export class EventStreamDecoder { } feed(chunk: string): void { + // Line endings between messages are extra blank lines (or the '\n' of a + // CRLF split after a delimiter), so they carry no content. + if (this.pending.length === 0) { + chunk = chunk.replace(LEADING_LINE_ENDINGS_REGEX, '') + } + // empty chunk has no meaningful content to process if (chunk === '') { return } - if (this.discardLeadingLF) { - this.discardLeadingLF = false - - if (chunk.charCodeAt(0) === LF) { - chunk = chunk.slice(1) - - // empty chunk has no meaningful content to process - if (chunk === '') { - return - } - } - } - const scan = this.tail + chunk + this.pending.push(chunk) + + MESSAGE_DELIMITER_REGEX.lastIndex = 0 + let match = MESSAGE_DELIMITER_REGEX.exec(scan) - if (!MESSAGE_DELIMITER_REGEX.test(scan)) { - this.pending.push(chunk) + if (match === null) { this.tail = scan.slice(-MAX_DELIMITER_OVERLAP) return } - this.pending.push(chunk) - const buffered = this.pending.length === 1 ? chunk : this.pending.join('') + const buffered = this.pending.join('') const offset = buffered.length - scan.length - const parts: string[] = [] let start = 0 - for (const match of scan.matchAll(MESSAGE_DELIMITER_GLOBAL_REGEX)) { + while (match !== null) { parts.push(buffered.slice(start, offset + match.index)) start = offset + match.index + match[0].length + match = MESSAGE_DELIMITER_REGEX.exec(scan) } const incomplete = buffered.slice(start) - this.pending.length = 0 + this.pending = incomplete === '' ? [] : [incomplete] this.tail = incomplete.slice(-MAX_DELIMITER_OVERLAP) - if (incomplete === '') { - this.discardLeadingLF = chunk.charCodeAt(chunk.length - 1) === CR - } - else { - this.pending.push(incomplete) - } - for (const encoded of parts) { this.onEvent(decodeEventStreamMessage(encoded)) } diff --git a/packages/fetch/src/event-stream.test.ts b/packages/fetch/src/event-stream.test.ts index a6be1e0..aeb03ca 100644 --- a/packages/fetch/src/event-stream.test.ts +++ b/packages/fetch/src/event-stream.test.ts @@ -92,6 +92,24 @@ describe('toAsyncIteratorObject', () => { await expect(stream.getReader().closed).resolves.toBe(undefined) }) + it('with extra blank lines', async () => { + const stream = new ReadableStream({ + async pull(controller) { + controller.enqueue('\n: ping\n\n\n') + 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.close() + }, + }).pipeThrough(new TextEncoderStream()) + + const generator = toAsyncIteratorObject(stream) + + expect(await generator.next()).toEqual({ done: false, value: { order: 1 } }) + expect(await generator.next()).toEqual({ done: false, value: { order: 2 } }) + expect(await generator.next()).toEqual({ done: true, value: undefined }) + }) + it('with empty stream', async () => { const generator = toAsyncIteratorObject(null) expect(generator).toSatisfy(isAsyncIteratorObject)