diff --git a/src/utils.ts b/src/utils.ts index d9e8a44..19c2ff4 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -16,6 +16,12 @@ export function writeFromReadableStreamDefaultReader( reader.cancel(error).catch(() => {}) } + if (writable.destroyed) { + // 'close' may already have been emitted, so cancel now instead of waiting for it. + cancel() + return reader.closed + } + writable.on('close', cancel) writable.on('error', cancel) ;(currentReadPromise ?? reader.read()).then(flow, handleStreamError) @@ -55,6 +61,7 @@ export function writeFromReadableStream(stream: ReadableStream, writ if (stream.locked) { throw new TypeError('ReadableStream is locked.') } else if (writable.destroyed) { + stream.cancel().catch(() => {}) return } diff --git a/test/listener.test.ts b/test/listener.test.ts index 1dd17d0..6fc5431 100644 --- a/test/listener.test.ts +++ b/test/listener.test.ts @@ -412,6 +412,38 @@ describe('Abort request', () => { await expect(req).rejects.toThrow() }) + it('should cancel the response body when the client disconnects before the response', async () => { + let resolveCancelled!: () => void + const cancelled = new Promise((r) => { + resolveCancelled = r + }) + + const server = createServer( + getRequestListener(async (req: Request) => { + await new Promise((r) => req.signal.addEventListener('abort', r)) + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: first\n\n')) + }, + cancel() { + resolveCancelled() + }, + }), + { headers: { 'content-type': 'text/event-stream' } } + ) + }) + ) + + await requestServer(server, { + method: 'GET', + path: '/abort', + signal: AbortSignal.timeout(50), + }).catch(() => {}) + + await withTimeout(cancelled, 'response body was not cancelled') + }) + it.each(['stream', 'session'] as const)( 'should emit an abort event when an HTTP/2 client closes the %s', async (target) => { diff --git a/test/utils.test.ts b/test/utils.test.ts index 4432bbf..c346a1e 100644 --- a/test/utils.test.ts +++ b/test/utils.test.ts @@ -2,6 +2,7 @@ import { Writable } from 'node:stream' import { buildOutgoingHttpHeaders, writeFromReadableStream, + writeFromReadableStreamDefaultReader, readWithoutBlocking, } from '../src/utils' @@ -130,6 +131,49 @@ describe('writeFromReadableStream', () => { expect(enqueueCalled).toBe(false) // enqueue should not be called expect(cancelCalled).toBe(true) // cancel should be called }) + + it('should cancel the stream if the client has already disconnected', async () => { + let cancelCalled = false + const stream = new ReadableStream({ + cancel() { + cancelCalled = true + }, + }) + + const writable = new Writable() + writable.destroy() + await new Promise((resolve) => writable.once('close', resolve)) + + await writeFromReadableStream(stream, writable) + + expect(cancelCalled).toBe(true) + }) +}) + +describe('writeFromReadableStreamDefaultReader', () => { + it('should cancel the reader if the client has already disconnected', async () => { + let cancelCalled = false + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('test')) + }, + cancel() { + cancelCalled = true + }, + }) + + const writable = new Writable({ + write(_chunk, _encoding, callback) { + callback() + }, + }) + writable.destroy() + await new Promise((resolve) => writable.once('close', resolve)) + + await writeFromReadableStreamDefaultReader(stream.getReader(), writable) + + expect(cancelCalled).toBe(true) + }) }) describe('readWithoutBlocking', () => {