fix(listener): Limit retries to a maximum of three.

src/listener.ts

@@ -131,9 +131,10 @@ const responseViaResponseObject = async (
     let done = false
     let currentReadPromise: Promise<ReadableStreamReadResult<Uint8Array>> | undefined = undefined
 
-    // In the case of synchronous responses, usually a maximum of two readings is done
-    for (let i = 0; i < 2; i++) {
-      currentReadPromise = reader.read()
+    // In the case of synchronous responses, usually a maximum of two (or three in special cases) readings is done
+    let maxReadCount = 2
+    for (let i = 0; i < maxReadCount; i++) {
+      currentReadPromise ||= reader.read()
       const chunk = await readWithoutBlocking(currentReadPromise).catch((e) => {
         console.error(e)
         done = true
@@ -143,7 +144,7 @@ const responseViaResponseObject = async (
           // XXX: In Node.js v24, some response bodies are not read all the way through until the next task queue,
           // so wait a moment and retry. (e.g. new Blob([new Uint8Array(contents)]) )
           await new Promise((resolve) => setTimeout(resolve))
-          i--
+          maxReadCount = 3
           continue
         }
 

test/server.test.ts

@@ -67,7 +67,7 @@
 
   it('Should return 200 response - POST /partially-consumed', async () => {
     const buffer = Buffer.alloc(1024 * 10) // large buffer
     const res = await new Promise<any>((resolve, reject) => {
       const req = request(server)
         .post('/partially-consumed')
         .set('Content-Length', buffer.length.toString())
@@ -88,7 +88,7 @@
 
   it('Should return 200 response - POST /partially-consumed-and-cancelled', async () => {
     const buffer = Buffer.alloc(1) // A large buffer will not make the test go far, so keep it small because it won't go far.
     const res = await new Promise<any>((resolve, reject) => {
       const req = request(server)
         .post('/partially-consumed-and-cancelled')
         .set('Content-Length', buffer.length.toString())
@@ -137,6 +137,8 @@
     const largeText = 'a'.repeat(1024 * 1024 * 10)
     let server: ServerType
     let resolveReadableStreamPromise: () => void
+    let resolveEventStreamPromise: () => void
+    let resolveEventStreamWithoutTransferEncodingPromise: () => void
     beforeAll(() => {
       const app = new Hono()
       app.use('*', async (c, next) => {
@@ -183,6 +185,43 @@
         })
         return new Response(stream)
       })
+      const eventStreamPromise = new Promise<void>((resolve) => {
+        resolveEventStreamPromise = resolve
+      })
+      app.get('/event-stream', () => {
+        const stream = new ReadableStream({
+          async start(controller) {
+            controller.enqueue('data: First!\n\n')
+            await eventStreamPromise
+            controller.enqueue('data: Second!\n\n')
+            controller.close()
+          },
+        })
+        return new Response(stream, {
+          headers: {
+            'content-type': 'text/event-stream',
+            'transfer-encoding': 'chunked',
+          },
+        })
+      })
+      const eventStreamWithoutTransferEncodingPromise = new Promise<void>((resolve) => {
+        resolveEventStreamWithoutTransferEncodingPromise = resolve
+      })
+      app.get('/event-stream-without-transfer-encoding', () => {
+        const stream = new ReadableStream({
+          async start(controller) {
+            controller.enqueue('data: First!\n\n')
+            await eventStreamWithoutTransferEncodingPromise
+            controller.enqueue('data: Second!\n\n')
+            controller.close()
+          },
+        })
+        return new Response(stream, {
+          headers: {
+            'content-type': 'text/event-stream',
+          },
+        })
+      })
       app.get('/buffer', () => {
         const response = new Response(Buffer.from('Hello Hono!'), {
           headers: { 'content-type': 'text/plain' },
@@ -256,6 +295,53 @@
       expect(expectedChunks.length).toBe(0) // all chunks are received
     })
 
+    it('Should return 200 response - GET /event-stream', async () => {
+      const expectedChunks = ['data: First!\n\n', 'data: Second!\n\n']
+      const resPromise = request(server)
+        .get('/event-stream')
+        .parse((res, fn) => {
+          // response header should be sent before sending data.
+          expect(res.headers['transfer-encoding']).toBe('chunked')
+          resolveEventStreamPromise()
+
+          res.on('data', (chunk) => {
+            const str = chunk.toString()
+            expect(str).toBe(expectedChunks.shift())
+          })
+          res.on('end', () => fn(null, ''))
+        })
+      await new Promise((resolve) => setTimeout(resolve, 100))
+      const res = await resPromise
+      expect(res.status).toBe(200)
+      expect(res.headers['content-type']).toMatch('text/event-stream')
+      expect(res.headers['content-length']).toBeUndefined()
+      expect(expectedChunks.length).toBe(0) // all chunks are received
+    })
+
+    it('Should return 200 response - GET /event-stream-without-transfer-encoding', async () => {
+      const expectedChunks = ['data: First!\n\n', 'data: Second!\n\n']
+      const resPromise = request(server)
+        .get('/event-stream-without-transfer-encoding')
+        .parse((res, fn) => {
+          res.on('data', (chunk) => {
+            const str = chunk.toString()
+            expect(str).toBe(expectedChunks.shift())
+
+            if (expectedChunks.length === 1) {
+              // receive first chunk
+              resolveEventStreamWithoutTransferEncodingPromise()
+            }
+          })
+          res.on('end', () => fn(null, ''))
+        })
+      await new Promise((resolve) => setTimeout(resolve, 100))
+      const res = await resPromise
+      expect(res.status).toBe(200)
+      expect(res.headers['content-type']).toMatch('text/event-stream')
+      expect(res.headers['content-length']).toBeUndefined()
+      expect(expectedChunks.length).toBe(0) // all chunks are received
+    })
+
     it('Should return 200 response - GET /buffer', async () => {
       const res = await request(server).get('/buffer')
       expect(res.status).toBe(200)