feat: add stream pipe response (#68)

Co-authored-by: Pooya Parsa <pyapar@gmail.com>

Author: stafyniaksacha

src/app.ts

@@ -3,7 +3,7 @@ import type { IncomingMessage, ServerResponse } from './types/node'
 import { lazyHandle, promisifyHandle } from './handle'
 import type { Handle, LazyHandle, Middleware, PHandle } from './handle'
 import { createError, sendError } from './error'
-import { send, MIMES } from './utils'
+import { send, sendStream, isStream, MIMES } from './utils'
 
 export interface Layer {
   route: string
@@ -114,6 +114,8 @@ export function createHandle (stack: Stack, options: AppOptions): PHandle {
       const type = typeof val
       if (type === 'string') {
         return send(res, val, MIMES.html)
+      } else if (isStream(val)) {
+        return sendStream(res, val)
       } else if (type === 'object' || type === 'boolean' || type === 'number' /* IS_JSON */) {
         if (val && val.buffer) {
           return send(res, val)

src/utils/response.ts

@@ -1,4 +1,5 @@
 import type { ServerResponse } from 'http'
+import { createError } from '../error'
 import { MIMES } from './consts'
 
 const defer = typeof setImmediate !== 'undefined' ? setImmediate : (fn: Function) => fn()
@@ -41,3 +42,15 @@ export function appendHeader (res: ServerResponse, name: string, value: string):
 
   res.setHeader(name, current.concat(value))
 }
+
+export function isStream (data: any) {
+  return typeof data === 'object' && typeof data.pipe === 'function' && typeof data.on === 'function'
+}
+
+export function sendStream (res: ServerResponse, data: any) {
+  return new Promise((resolve, reject) => {
+    data.pipe(res)
+    data.on('end', () => resolve(undefined))
+    data.on('error', (error: Error) => reject(createError(error)))
+  })
+}

test/app.test.ts

@@ -1,3 +1,4 @@
+import { Readable, Transform } from 'stream'
 import supertest, { SuperTest, Test } from 'supertest'
 import { describe, it, expect, beforeEach } from 'vitest'
 import { createApp, App } from '../src'
@@ -18,6 +19,48 @@ describe('app', () => {
     expect(res.body).toEqual({ url: '/' })
   })
 
+  it('can return Buffer directly', async () => {
+    app.use(() => Buffer.from('<h1>Hello world!</h1>', 'utf8'))
+    const res = await request.get('/')
+
+    expect(res.text).toBe('<h1>Hello world!</h1>')
+  })
+
+  it('can return Readable stream directly', async () => {
+    app.use(() => {
+      const readable = new Readable()
+      readable.push(Buffer.from('<h1>Hello world!</h1>', 'utf8'))
+      readable.push(null)
+      return readable
+    })
+    const res = await request.get('/')
+
+    expect(res.text).toBe('<h1>Hello world!</h1>')
+    expect(res.header['transfer-encoding']).toBe('chunked')
+  })
+
+  it('can return Readable stream that may throw', async () => {
+    app.use(() => {
+      const readable = new Readable()
+      const willThrow = new Transform({
+        transform (
+          _chunk,
+          _encoding,
+          callback
+        ) {
+          setTimeout(() => callback(new Error('test')), 0)
+        }
+      })
+      readable.push(Buffer.from('<h1>Hello world!</h1>', 'utf8'))
+      readable.push(null)
+
+      return readable.pipe(willThrow)
+    })
+    const res = await request.get('/')
+
+    expect(res.status).toBe(500)
+  })
+
   it('can return HTML directly', async () => {
     app.use(() => '<h1>Hello world!</h1>')
     const res = await request.get('/')