fix: adds steam clean up (#1910)

**Adds stream clean-up** in addition to nullish value clean-up.

- update from @fengmk2's identified issue
[here](#1893 (comment)).
   - #1893 (comment)
- Stackblitz display issue + fix
[here](https://stackblitz.com/~/github.com/yowainwright/koa-patch-ISSUE-1834_discussion_r2455395874?file=patch/koa/lib/response.js).
-
https://stackblitz.com/~/github.com/yowainwright/koa-patch-ISSUE-1834_discussion_r2455395874?file=patch/koa/lib/response.js

Author: yowainwright

__tests__/response/body.test.js

@@ -126,6 +126,111 @@ describe('res.body=', () => {
       res.body = body
       assert.strictEqual(body.listenerCount('error'), 0)
     })
+
+    it('should cleanup original stream when replaced by new stream', () => {
+      const res = response()
+      const stream1 = new Stream.PassThrough()
+      const stream2 = new Stream.PassThrough()
+
+      res.body = stream1
+      res.body = stream2
+
+      assert.strictEqual(stream1.destroyed, true)
+    })
+
+    it('should cleanup original stream when replaced by null', () => {
+      const res = response()
+      const stream = new Stream.PassThrough()
+
+      res.body = stream
+      res.body = null
+
+      assert.strictEqual(stream.destroyed, true)
+    })
+
+    it('should not throw unhandled errors when replacing failing stream', async () => {
+      const res = response()
+
+      const stream1 = new Stream.Readable({
+        read () {
+        }
+      })
+
+      const stream2 = new Stream.PassThrough()
+
+      res.body = stream1
+      res.body = stream2
+
+      await new Promise((resolve) => {
+        process.nextTick(() => {
+          stream1.emit('error', new Error('stream1 error'))
+          setTimeout(resolve, 10)
+        })
+      })
+    })
+
+    it('should handle multiple sequential stream replacements', () => {
+      const res = response()
+      const stream1 = new Stream.PassThrough()
+      const stream2 = new Stream.PassThrough()
+      const stream3 = new Stream.PassThrough()
+
+      res.body = stream1
+      res.body = stream2
+      res.body = stream3
+
+      assert.strictEqual(stream1.destroyed, true)
+      assert.strictEqual(stream2.destroyed, true)
+      assert.strictEqual(stream3.destroyed, false)
+    })
+
+    it('should handle four sequential stream replacements', () => {
+      const res = response()
+      const stream1 = new Stream.PassThrough()
+      const stream2 = new Stream.PassThrough()
+      const stream3 = new Stream.PassThrough()
+      const stream4 = new Stream.PassThrough()
+
+      res.body = stream1
+      res.body = stream2
+      res.body = stream3
+      res.body = stream4
+
+      assert.strictEqual(stream1.destroyed, true)
+      assert.strictEqual(stream2.destroyed, true)
+      assert.strictEqual(stream3.destroyed, true)
+      assert.strictEqual(stream4.destroyed, false)
+    })
+
+    it('should cleanup stream when replaced by string', () => {
+      const res = response()
+      const stream = new Stream.PassThrough()
+
+      res.body = stream
+      res.body = 'hello'
+
+      assert.strictEqual(stream.destroyed, true)
+    })
+
+    it('should cleanup stream when replaced by buffer', () => {
+      const res = response()
+      const stream = new Stream.PassThrough()
+
+      res.body = stream
+      res.body = Buffer.from('hello')
+
+      assert.strictEqual(stream.destroyed, true)
+    })
+
+    it('should cleanup stream when replaced by object', () => {
+      const res = response()
+      const stream = new Stream.PassThrough()
+
+      res.body = stream
+      res.body = { foo: 'bar' }
+
+      assert.strictEqual(stream.destroyed, true)
+    })
   })
 
   describe('when a buffer is given', () => {

lib/response.js

@@ -135,6 +135,14 @@ module.exports = {
   set body (val) {
     const original = this._body
     this._body = val
+
+    const cleanupPreviousStream = () => {
+      if (original && isStream(original)) {
+        original.once('error', () => {})
+        destroy(original)
+      }
+    }
+
     // no content
 
     if (val == null) {
@@ -149,13 +157,7 @@ module.exports = {
       this.remove('Content-Type')
       this.remove('Content-Length')
       this.remove('Transfer-Encoding')
-
-      const shouldDestroyOriginal = original && isStream(original)
-      if (shouldDestroyOriginal) {
-        // Ignore errors during cleanup to prevent unhandled exceptions when destroying the stream
-        original.once('error', () => {})
-        destroy(original)
-      }
+      cleanupPreviousStream()
       return
     }
 
@@ -169,22 +171,24 @@ module.exports = {
     if (typeof val === 'string') {
       if (setType) this.type = /^\s*</.test(val) ? 'html' : 'text'
       this.length = Buffer.byteLength(val)
+      cleanupPreviousStream()
       return
     }
 
     // buffer
     if (Buffer.isBuffer(val)) {
       if (setType) this.type = 'bin'
       this.length = val.length
+      cleanupPreviousStream()
       return
     }
 
     // stream
     if (isStream(val)) {
       onFinish(this.res, destroy.bind(null, val))
       if (original !== val) {
-        // overwriting
         if (original != null) this.remove('Content-Length')
+        cleanupPreviousStream()
       }
 
       if (setType) this.type = 'bin'
@@ -194,13 +198,15 @@ module.exports = {
     // ReadableStream
     if (val instanceof ReadableStream) {
       if (setType) this.type = 'bin'
+      cleanupPreviousStream()
       return
     }
 
     // blob
     if (val instanceof Blob) {
       if (setType) this.type = 'bin'
       this.length = val.size
+      cleanupPreviousStream()
       return
     }
 
@@ -212,13 +218,14 @@ module.exports = {
       for (const key of headers.keys()) {
         this.set(key, headers.get(key))
       }
-
+      cleanupPreviousStream()
       return
     }
 
     // json
     this.remove('Content-Length')
     if (!this.type || !/\bjson\b/i.test(this.type)) this.type = 'json'
+    cleanupPreviousStream()
   },
 
   /**