From d215db8379c01d7268068f3ebc2c9dc23183f75d Mon Sep 17 00:00:00 2001 From: Varun Chawla Date: Wed, 11 Mar 2026 21:25:13 -0700 Subject: [PATCH 1/2] feat: improve stream handling with async iterable support and better error handling Add support for async iterables (async generators) as response bodies, filter ERR_STREAM_PREMATURE_CLOSE errors from client disconnects in the stream pipeline, and add comprehensive tests for the new functionality. Closes #1864 --- __tests__/application/respond.test.js | 206 ++++++++++++++++++++++++++ __tests__/response/body.test.js | 45 ++++++ lib/application.js | 5 + lib/is-stream.js | 24 ++- lib/response.js | 9 ++ 5 files changed, 288 insertions(+), 1 deletion(-) diff --git a/__tests__/application/respond.test.js b/__tests__/application/respond.test.js index 79fe87f6f..ed8a9c578 100644 --- a/__tests__/application/respond.test.js +++ b/__tests__/application/respond.test.js @@ -943,6 +943,212 @@ describe('app.respond', () => { assert(streamDestroyed, 'Stream should be destroyed on client abort') }) + + it('should not emit error on premature close from client disconnect', async () => { + const app = new Koa() + const PassThrough = require('stream').PassThrough + const http = require('http') + + let errorEmitted = false + app.on('error', () => { + errorEmitted = true + }) + + app.use(ctx => { + const stream = new PassThrough() + ctx.body = stream + + setImmediate(() => { + stream.write('some data') + }) + }) + + const server = app.listen() + + await new Promise((resolve) => { + const req = http.request({ + port: server.address().port, + path: '/' + }) + + req.on('response', (res) => { + res.on('data', () => { + req.destroy() + setTimeout(() => { + server.close() + resolve() + }, 50) + }) + }) + + req.end() + }) + + assert.strictEqual(errorEmitted, false, 'ERR_STREAM_PREMATURE_CLOSE should not be emitted as an error') + }) + }) + + describe('when .body is an AsyncIterable', () => { + it('should respond with async generator content', async () => { + const app = new Koa() + + app.use(ctx => { + ctx.type = 'text/plain' + async function * generate () { + yield 'Hello ' + yield 'World' + } + ctx.body = generate() + }) + + const res = await request(app.callback()) + .get('/') + .expect(200) + + assert.strictEqual(res.text, 'Hello World') + }) + + it('should respond with async iterable content', async () => { + const app = new Koa() + + app.use(ctx => { + ctx.type = 'text/plain' + const iterable = { + [Symbol.asyncIterator] () { + let i = 0 + const chunks = ['chunk1', 'chunk2', 'chunk3'] + return { + async next () { + if (i < chunks.length) { + return { value: chunks[i++], done: false } + } + return { done: true } + } + } + } + } + ctx.body = iterable + }) + + const res = await request(app.callback()) + .get('/') + .expect(200) + + assert.strictEqual(res.text, 'chunk1chunk2chunk3') + }) + + it('should respond with async generator yielding buffers', async () => { + const app = new Koa() + + app.use(ctx => { + async function * generate () { + yield Buffer.from('Hello ') + yield Buffer.from('World') + } + ctx.body = generate() + }) + + const res = await request(app.callback()) + .get('/') + .expect(200) + + assert.deepStrictEqual(res.body, Buffer.from('Hello World')) + }) + + it('should default to octet-stream content type', async () => { + const app = new Koa() + + app.use(ctx => { + async function * generate () { + yield 'data' + } + ctx.body = generate() + }) + + return request(app.callback()) + .get('/') + .expect(200) + .expect('content-type', 'application/octet-stream') + }) + + it('should respect custom content type', async () => { + const app = new Koa() + + app.use(ctx => { + ctx.type = 'text/plain' + async function * generate () { + yield 'plain text content' + } + ctx.body = generate() + }) + + return request(app.callback()) + .get('/') + .expect(200) + .expect('content-type', 'text/plain; charset=utf-8') + }) + + it('should strip content-length when overwriting body with async iterable', async () => { + const app = new Koa() + + app.use(ctx => { + ctx.body = 'hello' + async function * generate () { + yield 'async content' + } + ctx.body = generate() + }) + + const res = await request(app.callback()) + .get('/') + .expect(200) + + assert.strictEqual(Object.prototype.hasOwnProperty.call(res.headers, 'content-length'), false) + assert.strictEqual(res.text, 'async content') + }) + + it('should handle async generator errors', async () => { + const app = new Koa() + + let errorCaught = false + app.once('error', err => { + assert.strictEqual(err.message, 'generator error') + errorCaught = true + }) + + app.use(ctx => { + async function * generate () { + yield 'start' + throw new Error('generator error') + } + ctx.body = generate() + }) + + await request(app.callback()) + .get('/') + .catch(() => {}) + + await new Promise(resolve => setTimeout(resolve, 50)) + assert(errorCaught, 'Error should have been caught') + }) + + it('should handle empty async generator', async () => { + const app = new Koa() + + app.use(ctx => { + ctx.type = 'text/plain' + async function * generate () { + // empty + } + ctx.body = generate() + }) + + const res = await request(app.callback()) + .get('/') + .expect(200) + + assert.strictEqual(res.text, '') + }) }) describe('when .body is an Object', () => { diff --git a/__tests__/response/body.test.js b/__tests__/response/body.test.js index c63cb4d75..bd48def44 100644 --- a/__tests__/response/body.test.js +++ b/__tests__/response/body.test.js @@ -304,6 +304,51 @@ describe('res.body=', () => { }) }) + describe('when an async iterable is given', () => { + it('should default to an octet stream', () => { + const res = response() + async function * generate () { + yield 'data' + } + res.body = generate() + assert.strictEqual('application/octet-stream', res.header['content-type']) + }) + + it('should not override Content-Type if already set', () => { + const res = response() + res.type = 'text/plain' + async function * generate () { + yield 'data' + } + res.body = generate() + assert.strictEqual('text/plain; charset=utf-8', res.header['content-type']) + }) + + it('should strip Content-Length when overwriting body', () => { + const res = response() + res.body = 'hello' + assert.strictEqual(5, res.header['content-length']) + async function * generate () { + yield 'data' + } + res.body = generate() + assert.strictEqual(res.header['content-length'], undefined) + }) + + it('should cleanup previous stream when replaced by async iterable', () => { + const res = response() + const stream = new Stream.PassThrough() + + res.body = stream + async function * generate () { + yield 'data' + } + res.body = generate() + + assert.strictEqual(stream.destroyed, true) + }) + }) + describe('when a response is given', () => { it('should set the status', () => { const res = response() diff --git a/lib/application.js b/lib/application.js index cd6504496..b29428cca 100644 --- a/lib/application.js +++ b/lib/application.js @@ -19,6 +19,7 @@ const request = require('./request') const response = require('./response') const context = require('./context') const isStream = require('./is-stream.js') +const { isAsyncIterable } = require('./is-stream.js') const only = require('./only.js') /** @typedef {typeof import ('./context') & { @@ -310,9 +311,13 @@ function respond (ctx) { else if (body instanceof ReadableStream) stream = Stream.Readable.from(body) else if (body instanceof Response) stream = Stream.Readable.from(body?.body || '') else if (isStream(body)) stream = body + else if (isAsyncIterable(body)) stream = Stream.Readable.from(body) if (stream) { return Stream.pipeline(stream, res, err => { + // Filter out premature close errors caused by client disconnects, + // as these are expected and not application errors. + if (err && err.code === 'ERR_STREAM_PREMATURE_CLOSE') return if (err && ctx.app.listenerCount('error')) ctx.onerror(err) }) } diff --git a/lib/is-stream.js b/lib/is-stream.js index b42368d9b..ab3dadbd8 100644 --- a/lib/is-stream.js +++ b/lib/is-stream.js @@ -4,7 +4,7 @@ const Stream = require('stream') // TODO: use a third party library for this -module.exports = (stream) => { +const isStream = (stream) => { return ( stream instanceof Stream || (stream !== null && @@ -18,3 +18,25 @@ module.exports = (stream) => { typeof stream.destroyed === 'boolean') ) } + +/** + * Check if `obj` is an async iterable (but not a string, Buffer, or Node.js stream). + * This enables support for async generators and other async iterables as response bodies. + * + * @param {*} obj + * @return {boolean} + */ +const isAsyncIterable = (obj) => { + return ( + obj !== null && + typeof obj === 'object' && + typeof obj[Symbol.asyncIterator] === 'function' && + !isStream(obj) && + !(obj instanceof ReadableStream) && + !(obj instanceof Blob) && + !(obj instanceof Response) + ) +} + +module.exports = isStream +module.exports.isAsyncIterable = isAsyncIterable diff --git a/lib/response.js b/lib/response.js index 846a8beed..b2d7d50e8 100644 --- a/lib/response.js +++ b/lib/response.js @@ -19,6 +19,7 @@ const vary = require('vary') const getType = require('mime-types').contentType const isStream = require('./is-stream.js') +const { isAsyncIterable } = require('./is-stream.js') const only = require('./only.js') /** @@ -225,6 +226,14 @@ module.exports = { return } + // async iterable (e.g. async generators) + if (isAsyncIterable(val)) { + if (original != null) this.remove('Content-Length') + if (setType) this.type = 'bin' + cleanupPreviousStream() + return + } + // json this.remove('Content-Length') if (!this.type || !/\bjson\b/i.test(this.type)) this.type = 'json' From 1351aa6994315d5aead74bba775a46f0c1741569 Mon Sep 17 00:00:00 2001 From: Varun Chawla Date: Wed, 11 Mar 2026 21:25:13 -0700 Subject: [PATCH 2/2] refactor: inline async iterable check, remove isAsyncIterable function The isAsyncIterable helper had redundant type exclusion checks since streams, ReadableStream, Blob, and Response are already handled by preceding branches in both the body setter and respond function. Inline the Symbol.asyncIterator check directly where needed. --- lib/application.js | 3 +-- lib/is-stream.js | 20 -------------------- lib/response.js | 3 +-- 3 files changed, 2 insertions(+), 24 deletions(-) diff --git a/lib/application.js b/lib/application.js index b29428cca..030ce620d 100644 --- a/lib/application.js +++ b/lib/application.js @@ -19,7 +19,6 @@ const request = require('./request') const response = require('./response') const context = require('./context') const isStream = require('./is-stream.js') -const { isAsyncIterable } = require('./is-stream.js') const only = require('./only.js') /** @typedef {typeof import ('./context') & { @@ -311,7 +310,7 @@ function respond (ctx) { else if (body instanceof ReadableStream) stream = Stream.Readable.from(body) else if (body instanceof Response) stream = Stream.Readable.from(body?.body || '') else if (isStream(body)) stream = body - else if (isAsyncIterable(body)) stream = Stream.Readable.from(body) + else if (typeof body[Symbol.asyncIterator] === 'function') stream = Stream.Readable.from(body) if (stream) { return Stream.pipeline(stream, res, err => { diff --git a/lib/is-stream.js b/lib/is-stream.js index ab3dadbd8..11540537c 100644 --- a/lib/is-stream.js +++ b/lib/is-stream.js @@ -19,24 +19,4 @@ const isStream = (stream) => { ) } -/** - * Check if `obj` is an async iterable (but not a string, Buffer, or Node.js stream). - * This enables support for async generators and other async iterables as response bodies. - * - * @param {*} obj - * @return {boolean} - */ -const isAsyncIterable = (obj) => { - return ( - obj !== null && - typeof obj === 'object' && - typeof obj[Symbol.asyncIterator] === 'function' && - !isStream(obj) && - !(obj instanceof ReadableStream) && - !(obj instanceof Blob) && - !(obj instanceof Response) - ) -} - module.exports = isStream -module.exports.isAsyncIterable = isAsyncIterable diff --git a/lib/response.js b/lib/response.js index b2d7d50e8..a8c5d089a 100644 --- a/lib/response.js +++ b/lib/response.js @@ -19,7 +19,6 @@ const vary = require('vary') const getType = require('mime-types').contentType const isStream = require('./is-stream.js') -const { isAsyncIterable } = require('./is-stream.js') const only = require('./only.js') /** @@ -227,7 +226,7 @@ module.exports = { } // async iterable (e.g. async generators) - if (isAsyncIterable(val)) { + if (typeof val[Symbol.asyncIterator] === 'function') { if (original != null) this.remove('Content-Length') if (setType) this.type = 'bin' cleanupPreviousStream()