fix(core): default the SSE event type to message, omit it when encoding

packages/aws-lambda/src/response.test.ts

@@ -132,7 +132,7 @@ describe('sendStandardResponse', () => {
       },
     })
 
-    expect(bodyOf(responseStream)).toBe(': \n\nevent: message\ndata: "foo"\n\nevent: message\ndata: "bar"\n\nevent: close\ndata: "baz"\n\n')
+    expect(bodyOf(responseStream)).toBe(': \n\ndata: "foo"\n\ndata: "bar"\n\nevent: close\ndata: "baz"\n\n')
     expect(responseStream.writableEnded).toBe(true)
   })
 

packages/core/src/event-stream/decoder.test.ts

@@ -62,6 +62,7 @@ describe('decodeEventStreamMessage', () => {
 
     // Per spec, only a single U+0020 SPACE is stripped — not tabs.
     expect(decodeEventStreamMessage('data:\thello\n\n')).toEqual({
+      event: 'message',
       data: '\thello',
     })
   })
@@ -76,10 +77,29 @@ describe('decodeEventStreamMessage', () => {
   })
 
   it('treats lines without a colon as fields with empty values', () => {
-    expect(decodeEventStreamMessage('data\n\n')).toEqual({ data: '' })
-    expect(decodeEventStreamMessage('data:\n\n')).toEqual({ data: '' })
-    expect(decodeEventStreamMessage('data: a\ndata:\ndata: b\n\n')).toEqual({ data: 'a\n\nb' })
-    expect(decodeEventStreamMessage('event\ndata: x\n\n')).toEqual({ event: '', data: 'x' })
+    expect(decodeEventStreamMessage('data\n\n')).toEqual({ event: 'message', data: '' })
+    expect(decodeEventStreamMessage('data:\n\n')).toEqual({ event: 'message', data: '' })
+    expect(decodeEventStreamMessage('data: a\ndata:\ndata: b\n\n')).toEqual({ event: 'message', data: 'a\n\nb' })
+    expect(decodeEventStreamMessage('event\ndata: x\n\n')).toEqual({ event: 'message', data: 'x' })
+  })
+
+  it('defaults the event type of an unnamed message with data to message', () => {
+    expect(decodeEventStreamMessage('data: x\n\n')).toStrictEqual({ event: 'message', data: 'x' })
+    expect(decodeEventStreamMessage('event:\ndata: x\n\n')).toStrictEqual({ event: 'message', data: 'x' })
+    expect(decodeEventStreamMessage('event: a\nevent:\ndata: x\n\n')).toStrictEqual({ event: 'message', data: 'x' })
+
+    // an empty data line still counts as data
+    expect(decodeEventStreamMessage('data:\n\n')).toStrictEqual({ event: 'message', data: '' })
+    expect(decodeEventStreamMessage('event:\ndata:\n\n')).toStrictEqual({ event: 'message', data: '' })
+
+    // named messages keep their event type, even without data
+    expect(decodeEventStreamMessage('event: close\n\n')).toStrictEqual({ event: 'close' })
+    expect(decodeEventStreamMessage('event: message\n\n')).toStrictEqual({ event: 'message' })
+
+    // unnamed messages without data get no event type
+    expect(decodeEventStreamMessage(': ping\n\n')).toStrictEqual({ comments: ['ping'] })
+    expect(decodeEventStreamMessage('id: 1\nretry: 10\n\n')).toStrictEqual({ id: '1', retry: 10 })
+    expect(decodeEventStreamMessage('event:\nid: 1\n\n')).toStrictEqual({ id: '1' })
   })
 
   it('ignores unknown and case-mismatched keys', () => {
@@ -161,7 +181,7 @@ describe('eventStreamDecoder', () => {
 
     it('ignores empty chunks', () => {
       expect(feedAll(['', 'data: hello', '', '\n\n', ''])).toEqual([
-        { data: 'hello' },
+        { event: 'message', data: 'hello' },
       ])
     })
 
@@ -221,8 +241,8 @@ describe('eventStreamDecoder', () => {
           const events = feedAll([stream.slice(0, split), stream.slice(split)])
 
           expect(events, `delimiter ${JSON.stringify(delimiter)} split at ${split}`).toEqual([
-            { data: 'first' },
-            { data: 'second' },
+            { event: 'message', data: 'first' },
+            { event: 'message', data: 'second' },
           ])
         }
       }
@@ -265,8 +285,8 @@ describe('eventStreamDecoder', () => {
       ])
 
       expect(events).toEqual([
-        { data: 'first' },
-        { data: 'second' },
+        { event: 'message', data: 'first' },
+        { event: 'message', data: 'second' },
       ])
     })
 
@@ -279,8 +299,8 @@ describe('eventStreamDecoder', () => {
       ])
 
       expect(events).toEqual([
-        { data: 'first' },
-        { data: 'second' },
+        { event: 'message', data: 'first' },
+        { event: 'message', data: 'second' },
       ])
     })
   })
@@ -318,7 +338,7 @@ describe('eventStreamDecoder', () => {
       expect(() => decoder.end()).toThrowError('Event Stream ended before complete')
 
       expect(events).toEqual([
-        { data: 'hello' },
+        { event: 'message', data: 'hello' },
       ])
     })
   })
@@ -390,7 +410,7 @@ describe('eventStreamDecoderStream', () => {
     const module = await import('./decoder')
 
     expect(module.EventStreamDecoderStream).toBeDefined()
-    expect(module.decodeEventStreamMessage('data: hello\n\n')).toEqual({ data: 'hello' })
+    expect(module.decodeEventStreamMessage('data: hello\n\n')).toEqual({ event: 'message', data: 'hello' })
   })
 
   it('on incomplete message', async () => {

packages/core/src/event-stream/decoder.ts

@@ -72,6 +72,18 @@ export function decodeEventStreamMessage(encoded: string): EventStreamMessage {
     }
   }
 
+  // Per spec, a message with no (or an empty) event name is a 'message' event if it has
+  // data, even an empty `data:` line. Without data it is never dispatched.
+  // https://html.spec.whatwg.org/multipage/server-sent-events.html#dispatchMessage
+  if (!message.event) {
+    if (message.data === undefined) {
+      delete message.event
+    }
+    else {
+      message.event = 'message'
+    }
+  }
+
   return message
 }
 

packages/core/src/event-stream/encoder.test.ts

@@ -132,22 +132,37 @@ describe('encodeEventStreamMessage', () => {
   })
 
   it('encodes fields in order: comments, event, retry, id, data', () => {
-    expect(encodeEventStreamMessage({ event: 'message', data: 'hello\nworld' }))
-      .toBe('event: message\ndata: hello\ndata: world\n\n')
+    expect(encodeEventStreamMessage({ event: 'close', data: 'hello\nworld' }))
+      .toBe('event: close\ndata: hello\ndata: world\n\n')
 
     expect(encodeEventStreamMessage({ event: 'message', id: '123', retry: 10000 }))
       .toBe('event: message\nretry: 10000\nid: 123\n\n')
 
-    expect(encodeEventStreamMessage({ event: 'message', id: '123', retry: 10000, comments: ['hello', 'world'] }))
-      .toBe(': hello\n: world\nevent: message\nretry: 10000\nid: 123\n\n')
+    expect(encodeEventStreamMessage({ event: 'close', id: '123', retry: 10000, data: 'hello', comments: ['hello', 'world'] }))
+      .toBe(': hello\n: world\nevent: close\nretry: 10000\nid: 123\ndata: hello\n\n')
+  })
+
+  it('omits the default message event type when the message has data', () => {
+    expect(encodeEventStreamMessage({ event: 'message', data: 'hello' })).toBe('data: hello\n\n')
+    expect(encodeEventStreamMessage({ event: 'message', data: '' })).toBe('data: \n\n')
+    expect(encodeEventStreamMessage({ event: 'message', id: '123', data: 'hello', comments: ['hi'] }))
+      .toBe(': hi\nid: 123\ndata: hello\n\n')
+
+    // kept without data, so it is not mistaken for a comment
+    expect(encodeEventStreamMessage({ event: 'message' })).toBe('event: message\n\n')
+    expect(encodeEventStreamMessage({ event: 'message', comments: ['hi'] })).toBe(': hi\nevent: message\n\n')
   })
 
   it('round-trips through decodeEventStreamMessage', () => {
     const messages: EventStreamMessage[] = [
       {},
-      { data: 'hello' },
-      { data: 'hello\nworld\n' },
+      { comments: ['ping'] },
+      { event: 'message' },
+      { event: 'close' },
+      { event: 'message', data: 'hello' },
+      { event: 'message', data: 'hello\nworld\n' },
       { event: 'message', data: 'hello', id: '123', retry: 10000, comments: ['hi'] },
+      { event: 'close', data: 'bye', id: '123' },
     ]
 
     for (const message of messages) {

packages/core/src/event-stream/encoder.ts

@@ -72,7 +72,9 @@ export function encodeEventStreamMessage(message: EventStreamMessage): string {
 
   output += encodeEventStreamMessageComments(message.comments)
 
-  if (message.event !== undefined) {
+  // 'message' is the default for a message with data. Without data, keep it
+  // so the message is not mistaken for a comment.
+  if (message.event !== undefined && (message.event !== 'message' || message.data === undefined)) {
     assertEventStreamMessageName(message.event)
 
     output += `event: ${message.event}\n`

packages/core/src/event-stream/types.ts

@@ -21,7 +21,7 @@ export interface EventMeta {
 
 export interface EventStreamMessage extends EventMeta {
   /**
-   * Event name (e.g., `message`, `error`).
+   * Event name (e.g., `message`, `error`). Defaults to `message` for a message with data.
    */
   event?: string | undefined
 

packages/fastify/src/response.test.ts

@@ -202,7 +202,7 @@ describe('sendStandardResponse', () => {
       'x-custom-header': 'custom-value',
     })
 
-    expect(res.text).toEqual(': \n\nevent: message\ndata: "foo"\n\nevent: message\ndata: "bar"\n\nevent: close\ndata: "baz"\n\n')
+    expect(res.text).toEqual(': \n\ndata: "foo"\n\ndata: "bar"\n\nevent: close\ndata: "baz"\n\n')
   })
 
   it('chunked (octet)', async ({ onTestFinished }) => {

packages/fetch/src/body.test.ts

@@ -453,7 +453,7 @@ describe('toFetchBody', () => {
     const reader = (body as ReadableStream).pipeThrough(new TextDecoderStream()).getReader()
 
     expect(await reader.read()).toEqual({ done: false, value: ': \n\n' })
-    expect(await reader.read()).toEqual({ done: false, value: 'event: message\ndata: 123\n\n' })
+    expect(await reader.read()).toEqual({ done: false, value: 'data: 123\n\n' })
     expect(await reader.read()).toEqual({ done: false, value: 'event: close\ndata: 456\n\n' })
   })
 

packages/fetch/src/event-stream.test.ts

@@ -110,6 +110,44 @@ describe('toAsyncIteratorObject', () => {
     expect(await generator.next()).toEqual({ done: true, value: undefined })
   })
 
+  it('treats messages without an event type as message events', async () => {
+    const stream = new ReadableStream<string>({
+      async pull(controller) {
+        controller.enqueue('data: {"order": 1}\nid: id-1\nretry: 10000\n\n')
+        controller.enqueue(': ping\nid: id-ignored\nretry: 20000\n\n')
+        controller.enqueue('event: message\ndata: {"order": 2}\n\n')
+        controller.enqueue('event:\ndata: {"order": 3}\nid: id-3\n\n')
+        controller.enqueue('data: 4\n\n')
+        controller.enqueue('data:\n\n')
+        controller.close()
+      },
+    }).pipeThrough(new TextEncoderStream())
+
+    const generator = toAsyncIteratorObject(stream)
+
+    expect(await generator.next()).toSatisfy(({ done, value }) => {
+      expect(done).toEqual(false)
+      expect(value).toStrictEqual({ order: 1 })
+      expect(getEventMeta(value)).toEqual({ id: 'id-1', retry: 10000 })
+
+      return true
+    })
+
+    expect(await generator.next()).toEqual({ done: false, value: { order: 2 } })
+
+    expect(await generator.next()).toSatisfy(({ done, value }) => {
+      expect(done).toEqual(false)
+      expect(value).toStrictEqual({ order: 3 })
+      expect(getEventMeta(value)).toEqual({ id: 'id-3' })
+
+      return true
+    })
+
+    expect(await generator.next()).toEqual({ done: false, value: 4 })
+    expect(await generator.next()).toEqual({ done: false, value: undefined })
+    expect(await generator.next()).toEqual({ done: true, value: undefined })
+  })
+
   it('with empty stream', async () => {
     const generator = toAsyncIteratorObject(null)
     expect(generator).toSatisfy(isAsyncIteratorObject)
@@ -228,8 +266,8 @@ describe('toEventStream', () => {
       .getReader()
 
     expect((await reader.read())).toEqual({ done: false, value: ': \n\n' })
-    expect((await reader.read())).toEqual({ done: false, value: 'event: message\nid: id-1\ndata: {"order":1}\n\n' })
-    expect((await reader.read())).toEqual({ done: false, value: 'event: message\nretry: 20000\ndata: {"order":2}\n\n' })
+    expect((await reader.read())).toEqual({ done: false, value: 'id: id-1\ndata: {"order":1}\n\n' })
+    expect((await reader.read())).toEqual({ done: false, value: 'retry: 20000\ndata: {"order":2}\n\n' })
     expect((await reader.read())).toEqual({ done: false, value: 'event: message\n\n' })
     expect((await reader.read())).toEqual({ done: false, value: 'event: close\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n' })
     expect((await reader.read())).toEqual({ done: true })
@@ -247,8 +285,8 @@ describe('toEventStream', () => {
       .getReader()
 
     expect((await reader.read())).toEqual({ done: false, value: ': \n\n' })
-    expect((await reader.read())).toEqual({ done: false, value: 'event: message\nid: id-1\ndata: {"order":1}\n\n' })
-    expect((await reader.read())).toEqual({ done: false, value: 'event: message\nretry: 20000\ndata: {"order":2}\n\n' })
+    expect((await reader.read())).toEqual({ done: false, value: 'id: id-1\ndata: {"order":1}\n\n' })
+    expect((await reader.read())).toEqual({ done: false, value: 'retry: 20000\ndata: {"order":2}\n\n' })
     expect((await reader.read())).toEqual({ done: false, value: 'event: message\n\n' })
     expect((await reader.read())).toEqual({ done: false, value: 'event: close\n\n' })
     expect((await reader.read())).toEqual({ done: true })
@@ -267,8 +305,8 @@ describe('toEventStream', () => {
       .getReader()
 
     expect((await reader.read())).toEqual({ done: false, value: ': \n\n' })
-    expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
+    expect((await reader.read()).value).toEqual('id: id-1\ndata: {"order":1}\n\n')
+    expect((await reader.read()).value).toEqual('retry: 20000\ndata: {"order":2}\n\n')
     expect((await reader.read()).value).toEqual('event: message\n\n')
     await expect(reader.read()).rejects.toThrow('order-4')
   })
@@ -286,8 +324,8 @@ describe('toEventStream', () => {
       .getReader()
 
     expect((await reader.read())).toEqual({ done: false, value: ': \n\n' })
-    expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
+    expect((await reader.read()).value).toEqual('id: id-1\ndata: {"order":1}\n\n')
+    expect((await reader.read()).value).toEqual('retry: 20000\ndata: {"order":2}\n\n')
     expect((await reader.read()).value).toEqual('event: message\n\n')
     expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n')
     expect((await reader.read()).done).toEqual(true)
@@ -476,7 +514,7 @@ describe('toEventStream', () => {
       ])
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello"\n\n' }),
         vi.advanceTimersByTimeAsync(20),
       ])
 
@@ -491,7 +529,7 @@ describe('toEventStream', () => {
       ])
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello"\n\n' }),
         vi.advanceTimersByTimeAsync(20),
       ])
 
@@ -517,12 +555,12 @@ describe('toEventStream', () => {
         .getReader()
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello1"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello1"\n\n' }),
         vi.advanceTimersByTimeAsync(100),
       ])
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello2"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello2"\n\n' }),
         vi.advanceTimersByTimeAsync(100),
       ])
 
@@ -551,7 +589,7 @@ describe('toEventStream', () => {
       await expect(reader.read()).resolves.toEqual({ done: false, value: ': stream-started\n\n' })
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello"\n\n' }),
         vi.advanceTimersByTimeAsync(50),
       ])
 
@@ -575,7 +613,7 @@ describe('toEventStream', () => {
         .getReader()
 
       await Promise.all([
-        expect(reader.read()).resolves.toEqual({ done: false, value: 'event: message\ndata: "hello"\n\n' }),
+        expect(reader.read()).resolves.toEqual({ done: false, value: 'data: "hello"\n\n' }),
         vi.advanceTimersByTimeAsync(50),
       ])
 

packages/node/src/body.test.ts

@@ -638,7 +638,7 @@ describe('toNodeHttpBody', () => {
     const reader = Readable.toWeb((body as Readable)).pipeThrough(new TextDecoderStream()).getReader()
 
     expect(await reader.read()).toEqual({ done: false, value: ': \n\n' })
-    expect(await reader.read()).toEqual({ done: false, value: 'event: message\ndata: 123\n\n' })
+    expect(await reader.read()).toEqual({ done: false, value: 'data: 123\n\n' })
     expect(await reader.read()).toEqual({ done: false, value: 'event: close\ndata: 456\n\n' })
     expect(await reader.read()).toEqual({ done: true })
   })

packages/node/src/event-stream.test.ts

@@ -47,9 +47,9 @@ it('toEventStream', async () => {
     .getReader()
 
   expect((await reader.read())).toEqual({ done: false, value: ': \n\n' })
-  expect((await reader.read())).toEqual({ done: false, value: 'event: message\ndata: 1\n\n' })
-  expect((await reader.read())).toEqual({ done: false, value: 'event: message\ndata: 2\n\n' })
-  expect((await reader.read())).toEqual({ done: false, value: 'event: message\ndata: 3\n\n' })
+  expect((await reader.read())).toEqual({ done: false, value: 'data: 1\n\n' })
+  expect((await reader.read())).toEqual({ done: false, value: 'data: 2\n\n' })
+  expect((await reader.read())).toEqual({ done: false, value: 'data: 3\n\n' })
   expect((await reader.read())).toEqual({ done: false, value: 'event: close\n\n' })
   expect((await reader.read())).toEqual({ done: true, value: undefined })