diff --git a/packages/aws-lambda/src/response.test.ts b/packages/aws-lambda/src/response.test.ts index b374b2c3..a18efcd1 100644 --- a/packages/aws-lambda/src/response.test.ts +++ b/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) }) diff --git a/packages/core/src/event-stream/decoder.test.ts b/packages/core/src/event-stream/decoder.test.ts index 6369a1e9..a7945b90 100644 --- a/packages/core/src/event-stream/decoder.test.ts +++ b/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 () => { diff --git a/packages/core/src/event-stream/decoder.ts b/packages/core/src/event-stream/decoder.ts index 1ac09d48..70175e25 100644 --- a/packages/core/src/event-stream/decoder.ts +++ b/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 } diff --git a/packages/core/src/event-stream/encoder.test.ts b/packages/core/src/event-stream/encoder.test.ts index b9f6bdb6..e05addc2 100644 --- a/packages/core/src/event-stream/encoder.test.ts +++ b/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) { diff --git a/packages/core/src/event-stream/encoder.ts b/packages/core/src/event-stream/encoder.ts index 3a818131..0b4dc5df 100644 --- a/packages/core/src/event-stream/encoder.ts +++ b/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` diff --git a/packages/core/src/event-stream/types.ts b/packages/core/src/event-stream/types.ts index 94298c10..5dedddfd 100644 --- a/packages/core/src/event-stream/types.ts +++ b/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 diff --git a/packages/fastify/src/response.test.ts b/packages/fastify/src/response.test.ts index 0405ba3f..97e0425f 100644 --- a/packages/fastify/src/response.test.ts +++ b/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 }) => { diff --git a/packages/fetch/src/body.test.ts b/packages/fetch/src/body.test.ts index 87b086d8..03192c2a 100644 --- a/packages/fetch/src/body.test.ts +++ b/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' }) }) diff --git a/packages/fetch/src/event-stream.test.ts b/packages/fetch/src/event-stream.test.ts index dab0e27e..9bfbd8bd 100644 --- a/packages/fetch/src/event-stream.test.ts +++ b/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({ + 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), ]) diff --git a/packages/node/src/body.test.ts b/packages/node/src/body.test.ts index db10139f..394b8d1b 100644 --- a/packages/node/src/body.test.ts +++ b/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 }) }) diff --git a/packages/node/src/event-stream.test.ts b/packages/node/src/event-stream.test.ts index 2c83eb57..477a7280 100644 --- a/packages/node/src/event-stream.test.ts +++ b/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 }) diff --git a/packages/node/src/response.test.ts b/packages/node/src/response.test.ts index 209230b4..a8676ea3 100644 --- a/packages/node/src/response.test.ts +++ b/packages/node/src/response.test.ts @@ -155,7 +155,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 () => { @@ -345,8 +345,8 @@ describe('sendStandardResponse', () => { expect(chunks).toEqual([ Buffer.from(': \n\n'), - Buffer.from('event: message\ndata: 1\n\n'), - Buffer.from('event: message\ndata: 2\n\n'), + Buffer.from('data: 1\n\n'), + Buffer.from('data: 2\n\n'), ]) expect(responseStream.closed).toBe(false) @@ -402,8 +402,8 @@ describe('sendStandardResponse', () => { expect(chunks).toEqual([ Buffer.from(': \n\n'), - Buffer.from('event: message\ndata: 1\n\n'), - Buffer.from('event: message\ndata: 2\n\n'), + Buffer.from('data: 1\n\n'), + Buffer.from('data: 2\n\n'), ]) expect(responseStream.closed).toBe(false)