feat: support custom streams

closes #1825

__tests__/response/body.test.js

@@ -2,6 +2,7 @@
 
 const { describe, it } = require('node:test')
 const response = require('../../test-helpers/context').response
+const CustomStream = require('../../test-helpers/stream')
 const assert = require('assert')
 const fs = require('fs')
 const Stream = require('stream')
@@ -110,6 +111,12 @@ describe('res.body=', () => {
       assert.strictEqual('application/octet-stream', res.header['content-type'])
     })
 
+    it('should support custom stream', () => {
+      const res = response()
+      res.body = new CustomStream.Readable()
+      assert.strictEqual('application/octet-stream', res.header['content-type'])
+    })
+
     it('should add error handler to the stream, but only once', () => {
       const res = response()
       const body = new Stream.PassThrough()

lib/application.js

@@ -16,6 +16,7 @@ const Emitter = require('events')
 const util = require('util')
 const Stream = require('stream')
 const http = require('http')
+const isStream = require('./is-stream.js')
 const only = require('./only.js')
 const { HttpError } = require('http-errors')
 
@@ -304,10 +305,10 @@ function respond (ctx) {
 
   if (Buffer.isBuffer(body)) return res.end(body)
   if (typeof body === 'string') return res.end(body)
-  if (body instanceof Stream) return body.pipe(res)
   if (body instanceof Blob) return Stream.Readable.from(body.stream()).pipe(res)
   if (body instanceof ReadableStream) return Stream.Readable.from(body).pipe(res)
   if (body instanceof Response) return Stream.Readable.from(body?.body || '').pipe(res)
+  if (isStream(body)) return body.pipe(res)
 
   // body: json
   body = JSON.stringify(body)

lib/is-stream.js

@@ -0,0 +1,18 @@
+'use strict'
+
+const Stream = require('stream')
+
+module.exports = (stream) => {
+  return (
+    stream instanceof Stream ||
+    (stream !== null &&
+      typeof stream === 'object' &&
+      !!stream.readable &&
+      typeof stream.pipe === 'function' &&
+      typeof stream.read === 'function' &&
+      typeof stream.readable === 'boolean' &&
+      typeof stream.readableObjectMode === 'boolean' &&
+      typeof stream.destroy === 'function' &&
+      typeof stream.destroyed === 'boolean')
+  )
+}

lib/response.js

@@ -14,10 +14,10 @@ const destroy = require('destroy')
 const assert = require('assert')
 const extname = require('path').extname
 const vary = require('vary')
+const isStream = require('./is-stream.js')
 const only = require('./only.js')
 const util = require('util')
 const encodeUrl = require('encodeurl')
-const Stream = require('stream')
 const deprecate = require('depd')('koa')
 
 /**
@@ -172,7 +172,7 @@ module.exports = {
     }
 
     // stream
-    if (val instanceof Stream) {
+    if (isStream(val)) {
       onFinish(this.res, destroy.bind(null, val))
       if (original !== val) {
         val.once('error', err => this.ctx.onerror(err))
@@ -240,7 +240,7 @@ module.exports = {
     }
 
     const { body } = this
-    if (!body || body instanceof Stream) return undefined
+    if (!body || isStream(body)) return undefined
     if (typeof body === 'string') return Buffer.byteLength(body)
     if (Buffer.isBuffer(body)) return body.length
     return Buffer.byteLength(JSON.stringify(body))

test-helpers/stream.js

@@ -0,0 +1,24 @@
+'use strict'
+
+const { EventEmitter } = require('events')
+
+class Readable extends EventEmitter {
+  pipe () {}
+  read () {}
+  destroy () {}
+  get readable () {
+    return true
+  }
+
+  get readableObjectMode () {
+    return false
+  }
+
+  get destroyed () {
+    return false
+  }
+}
+
+module.exports = {
+  Readable
+}