fix(core): ignore extra blank lines in event stream decoder

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' },
+      ])
+    })
   })
 })
 

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,69 +69,56 @@ 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,
   ) {
   }
 
   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))
     }

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<string>({
+      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)