Readable streams 309

index.js

@@ -5,12 +5,12 @@ const { inherits, format } = require('node:util')
 
 const fp = require('fastify-plugin')
 const encodingNegotiator = require('@fastify/accept-negotiator')
-const pump = require('pump')
 const mimedb = require('mime-db')
 const peek = require('peek-stream')
 const { Minipass } = require('minipass')
 const pumpify = require('pumpify')
 const { Readable } = require('readable-stream')
+const { pipeline } = require('node:stream')
 
 const { isStream, isGzip, isDeflate, intoAsyncIterator } = require('./lib/utils')
 
@@ -267,12 +267,12 @@ function buildRouteCompress (fastify, params, routeOptions, decorateOnly) {
         encoding === undefined
           ? reply.removeHeader('Content-Encoding')
           : reply.header('Content-Encoding', 'identity')
-        pump(stream, payload = unzipStream(params.uncompressStream), onEnd.bind(reply))
+        pipeline(stream, payload = unzipStream(params.uncompressStream), onEnd.bind(reply))
       }
       return next(null, payload)
     }
 
-    if (typeof payload.pipe !== 'function') {
+    if (typeof payload.pipe !== 'function' && !(payload instanceof ReadableStream)) {
       if (Buffer.byteLength(payload) < params.threshold) {
         return next()
       }
@@ -286,7 +286,7 @@ function buildRouteCompress (fastify, params, routeOptions, decorateOnly) {
     }
 
     stream = zipStream(params.compressStream, encoding)
-    pump(payload, stream, onEnd.bind(reply))
+    pipeline(payload, stream, onEnd.bind(reply))
     next(null, stream)
   }
 }
@@ -348,7 +348,7 @@ function buildRouteDecompress (fastify, params, routeOptions) {
     raw.on('data', trackEncodedLength.bind(decompresser))
     raw.on('end', removeEncodedLengthTracking)
 
-    next(null, pump(raw, decompresser))
+    next(null, pipeline(raw, decompresser))
   }
 }
 
@@ -385,18 +385,17 @@ function compress (params) {
         encoding === undefined
           ? this.removeHeader('Content-Encoding')
           : this.header('Content-Encoding', 'identity')
-        pump(stream, payload = unzipStream(params.uncompressStream), onEnd.bind(this))
+        pipeline(stream, payload = unzipStream(params.uncompressStream), onEnd.bind(this))
       }
       return this.send(payload)
     }
 
-    if (typeof payload.pipe !== 'function') {
+    if (typeof payload.pipe !== 'function' && !(payload instanceof ReadableStream)) {
+
       if (!Buffer.isBuffer(payload) && typeof payload !== 'string') {
         payload = this.serialize(payload)
       }
-    }
 
-    if (typeof payload.pipe !== 'function') {
       if (Buffer.byteLength(payload) < params.threshold) {
         return this.send(payload)
       }
@@ -410,7 +409,7 @@ function compress (params) {
     }
 
     stream = zipStream(params.compressStream, encoding)
-    pump(payload, stream, onEnd.bind(this))
+    pipeline(payload, stream, onEnd.bind(this))
     this.send(stream)
   }
 }

test/regression/issue-309.test.js

@@ -0,0 +1,41 @@
+'use strict'
+
+const { test } = require('node:test')
+const Fastify = require('fastify')
+const fastifyCompress = require('../..')
+const zlib = require('node:zlib')
+
+test('it should handle ReadableStream', async (t) => {
+    const expectedData = {
+        data: 'to compress'
+    }
+
+    const fastify = new Fastify()
+    t.after(() => fastify.close())
+    fastify.register(async (instance, opts) => {
+        await fastify.register(fastifyCompress, {
+            threshold: 8
+        })
+        instance.get('/broken', async (req, reply) => {
+            const stream = ReadableStream.from(JSON.stringify(expectedData))
+            reply
+                .type('application/json')
+                .send(stream)
+            return reply
+        })
+      })
+    try {
+        const response = await fastify.inject({
+            method: 'GET',
+            url: '/broken',
+            headers: {
+              'accept-encoding': 'deflate'
+            }
+        })
+        const payload = zlib.inflateSync(response.rawPayload)
+
+        t.assert.deepStrictEqual(JSON.parse(payload.toString('utf-8')), expectedData)
+    } catch (e) {
+        t.assert.fail(e)
+    }
+})