fix: stray reconnect after close

Author: rexxars

src/client.ts

@@ -90,8 +90,8 @@ export function createEventSource(
         request = null
 
         // We expect abort errors when the user manually calls `close()` - ignore those
-        if (err.name !== 'AbortError' && err.type !== 'aborted') {
-          throw err
+        if (err.name === 'AbortError' || err.type === 'aborted') {
+          return
         }
 
         scheduleReconnect()

test/server.ts

@@ -8,6 +8,7 @@ import esbuild from 'esbuild'
 import {unicodeLines} from './fixtures.js'
 
 const isDeno = typeof globalThis.Deno !== 'undefined'
+const idedListeners = new Map<string, Set<ServerResponse>>()
 
 export function getServer(port: number): Promise<Server> {
   return new Promise((resolve, reject) => {
@@ -23,12 +24,15 @@ function onRequest(req: IncomingMessage, res: ServerResponse) {
     res.socket.setNoDelay(true)
   }
 
-  switch (req.url) {
+  const path = new URL(req.url || '/', 'http://localhost').pathname
+  switch (path) {
     // Server-Sent Event endpoints
     case '/':
       return writeDefault(req, res)
     case '/counter':
       return writeCounter(req, res)
+    case '/identified':
+      return writeIdentifiedListeners(req, res)
     case '/end-after-one':
       return writeOne(req, res)
     case '/slow-connect':
@@ -109,6 +113,42 @@ async function writeCounter(req: IncomingMessage, res: ServerResponse) {
   res.end()
 }
 
+function writeIdentifiedListeners(req: IncomingMessage, res: ServerResponse) {
+  const url = new URL(req.url || '/', 'http://localhost')
+  const id = url.searchParams.get('id')
+  if (!id) {
+    res.writeHead(400, {
+      'Content-Type': 'application/json',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive',
+    })
+    tryWrite(res, JSON.stringify({error: 'Missing "id" query parameter'}))
+    res.end()
+    return
+  }
+
+  if ((req.headers.accept || '').includes('text/event-stream')) {
+    const current = (idedListeners.get(id) || new Set()).add(res)
+    idedListeners.set(id, current)
+    res.writeHead(200, {
+      'Content-Type': 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive',
+    })
+    tryWrite(res, formatEvent({data: '', retry: 250}))
+    tryWrite(res, formatEvent({data: `${idedListeners.size}`}))
+    res.on('close', () => (idedListeners.get(id) || new Set()).delete(res))
+    return
+  }
+
+  res.writeHead(200, {
+    'Content-Type': 'application/json',
+    'Cache-Control': 'no-cache',
+  })
+  tryWrite(res, JSON.stringify({numListeners: (idedListeners.get(id) || new Set()).size}))
+  res.end()
+}
+
 function writeOne(req: IncomingMessage, res: ServerResponse) {
   const last = getLastEventId(req)
   res.writeHead(last ? 204 : 200, {

test/tests.ts

@@ -118,6 +118,38 @@ export function registerTests(options: {
     await deferClose(es)
   })
 
+  test('will not reconnect after explicit close()', async () => {
+    const request = fetch || globalThis.fetch
+    const onMessage = getCallCounter()
+    const onDisconnect = getCallCounter()
+    const url = `${baseUrl}:${port}/identified?id=explicit-close-no-reconnect`
+    const es = createEventSource({
+      url,
+      fetch,
+      onMessage,
+      onDisconnect,
+    })
+
+    // Should receive a message containing the number of listeners on the given ID
+    await onMessage.waitForCallCount(1)
+    expect(onMessage.lastCall.lastArg).toMatchObject({data: '1'})
+    expect(es.readyState).toBe(OPEN) // Open (connected)
+
+    // Explicitly disconnect. Should normally reconnect within ~250ms (server sends retry: 250)
+    // but we'll close it before that happens
+    es.close()
+    expect(es.readyState).toBe(CLOSED)
+    expect(onMessage.callCount).toBe(1)
+
+    // After 500 ms, there should be no clients connected to the given ID
+    await new Promise((resolve) => setTimeout(resolve, 500))
+    expect(await request(url).then((res) => res.json())).toMatchObject({numListeners: 0})
+
+    // Wait another 500 ms, just to be sure there are no slow reconnects
+    await new Promise((resolve) => setTimeout(resolve, 500))
+    expect(await request(url).then((res) => res.json())).toMatchObject({numListeners: 0})
+  })
+
   test('can use async iterator, reconnects transparently', async () => {
     const onDisconnect = getCallCounter()
     const url = `${baseUrl}:${port}/counter`