fix: stop dispatching messages after close

.changeset/quiet-stream-close.md

@@ -0,0 +1,5 @@
+---
+'eventsource': patch
+---
+
+Stop dispatching buffered messages when an event listener closes the connection.

src/EventSource.ts

@@ -601,6 +601,11 @@ class EventSourceImpl extends EventTarget implements EventSource {
    * @internal
    */
   #onEvent = (event: EventSourceMessage) => {
+    // A listener can close the connection while the parser is still processing this chunk.
+    if (this.#readyState === this.CLOSED) {
+      return
+    }
+
     const origin = this.#redirectUrl ? this.#redirectUrl.origin : this.#url.origin
     // [spec] The `lastEventId` attribute is the last event ID string of the event
     // source, i.e. the persisted buffer (`#lastEventId`) - not the current event's `id`.

test/client.test.ts

@@ -44,6 +44,62 @@ const xOriginRedirectTest = suite === 'happy-dom' ? test.fails : test
  */
 const onHandlerTest = suite === 'workerd' ? test.fails : test
 
+const bufferedEventTypes = [
+  {name: 'nameless', eventType: ''},
+  {name: 'message', eventType: 'message'},
+  {name: 'notice', eventType: 'notice'},
+]
+
+test.each(bufferedEventTypes)(
+  'stops dispatching buffered $name events when a listener closes the connection',
+  async ({eventType}) => {
+    const seen: string[] = []
+    const onMessage = getCallCounter({name: 'first message'})
+    const es = new OurEventSource(`${serverUrl}/message-burst?event=${eventType}`, {
+      async fetch(url, init) {
+        const response = await request(url, init)
+        // Keep the real HTTP exchange, but make the chunk boundary deterministic.
+        const body = await response.arrayBuffer()
+        return new Response(body, {status: response.status, headers: response.headers})
+      },
+    })
+
+    es.addEventListener(eventType || 'message', (event) => {
+      seen.push(event.data)
+      es.close()
+      onMessage.listener(event)
+    })
+
+    try {
+      await onMessage.waitForCallCount(1)
+      expect(seen).toEqual(['first'])
+      expect(es.readyState).toBe(OurEventSource.CLOSED)
+    } finally {
+      es.close()
+    }
+  },
+)
+
+test.each(bufferedEventTypes)(
+  'dispatches every buffered $name event while open',
+  async ({eventType}) => {
+    const seen: string[] = []
+    const onMessage = getCallCounter({name: 'messages'})
+    const es = new OurEventSource(`${serverUrl}/message-burst?event=${eventType}`, esInit)
+    es.addEventListener(eventType || 'message', (event) => {
+      seen.push(event.data)
+      onMessage.listener(event)
+    })
+
+    try {
+      await onMessage.waitForCallCount(3)
+      expect(seen).toEqual(['first', 'second', 'third'])
+    } finally {
+      es.close()
+    }
+  },
+)
+
 test('can connect, receive message, manually disconnect', async () => {
   const onMessage = getCallCounter({name: 'onMessage'})
   const es = new OurEventSource(new URL(`${serverUrl}/`))

test/helpers/server.ts

@@ -61,6 +61,8 @@ export function handleRequest(
       return writeCounter(req, res)
     case '/mixed-ids':
       return writeMixedIds(req, res)
+    case '/message-burst':
+      return writeMessageBurst(req, res)
     case '/id-only':
       return writeIdOnly(req, res)
     case '/identified':
@@ -145,6 +147,12 @@ async function writeCounter(req: IncomingMessage, res: ServerResponse) {
   res.end()
 }
 
+function writeMessageBurst(req: IncomingMessage, res: ServerResponse) {
+  const event = new URL(req.url || '/', 'http://localhost').searchParams.get('event') ?? 'message'
+  res.writeHead(200, {'Content-Type': 'text/event-stream'})
+  res.end(['first', 'second', 'third'].map((data) => encode({event, data})).join(''))
+}
+
 /**
  * Writes two messages: one with an `id` field, then one without. Per the spec, the second
  * event's `lastEventId` must still be `'1'`: the last event ID buffer is only updated by an