Attempt to use node:stream pipeline instead of pump - failing tests

index.js

@@ -11,6 +11,7 @@ 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,15 +268,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 (payload instanceof ReadableStream) {
-      payload = require('node:stream').Readable.fromWeb(payload)
-    }
 
-    if (typeof payload.pipe !== 'function') {
+    if (typeof payload.pipe !== 'function' && !(payload instanceof ReadableStream)) {
       if (Buffer.byteLength(payload) < params.threshold) {
         return next()
       }
@@ -289,7 +287,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)
   }
 }
@@ -351,7 +349,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))
   }
 }
 
@@ -388,16 +386,12 @@ 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 (payload instanceof ReadableStream) {
-      payload = require('node:stream').Readable.fromWeb(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)
       }
@@ -417,7 +411,7 @@ function compress (params) {
     }
 
     stream = zipStream(params.compressStream, encoding)
-    pump(payload, stream, onEnd.bind(this))
+    pipeline(payload, stream, onEnd.bind(this))
     this.send(stream)
   }
 }