stream: re-use legacy destroyer

PR-URL: https://github.com/nodejs/node/pull/31316
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>

3 files changed, 18 insertions, 20 deletions

diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index 6d798ec2ff..c4c220a5af 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -11,6 +11,7 @@ const {
 } = primordials;
 
 const finished = require('internal/streams/end-of-stream');
+const destroyImpl = require('internal/streams/destroy');
 
 const kLastResolve = Symbol('lastResolve');
 const kLastReject = Symbol('lastReject');
@@ -22,15 +23,6 @@ const kStream = Symbol('stream');
 
 let Readable;
 
-function destroy(stream, err) {
-  // request.destroy just do .end - .abort is what we want
-  if (typeof stream.abort === 'function') return stream.abort();
-  if (stream.req &&
-      typeof stream.req.abort === 'function') return stream.req.abort();
-  if (typeof stream.destroy === 'function') return stream.destroy(err);
-  if (typeof stream.close === 'function') return stream.close();
-}
-
 function createIterResult(value, done) {
   return { value, done };
 }
@@ -92,7 +84,7 @@ function finish(self, err) {
         resolve(createIterResult(undefined, true));
       }
     });
-    destroy(stream, err);
+    destroyImpl.destroyer(stream, err);
   });
 }
 
@@ -174,7 +166,7 @@ const createReadableStreamAsyncIterator = (stream) => {
 
     const src = stream;
     stream = new Readable({ objectMode: true }).wrap(src);
-    finished(stream, (err) => destroy(src, err));
+    finished(stream, (err) => destroyImpl.destroyer(src, err));
   }
 
   const iterator = ObjectCreate(ReadableStreamAsyncIteratorPrototype, {
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index b80fb56a6b..8837af2d71 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -151,8 +151,21 @@ function errorOrDestroy(stream, err, sync) {
   }
 }
 
+function isRequest(stream) {
+  return stream && stream.setHeader && typeof stream.abort === 'function';
+}
+
+// Normalize destroy for legacy.
+function destroyer(stream, err) {
+  // request.destroy just do .end - .abort is what we want
+  if (isRequest(stream)) return stream.abort();
+  if (isRequest(stream.req)) return stream.req.abort();
+  if (typeof stream.destroy === 'function') return stream.destroy(err);
+  if (typeof stream.close === 'function') return stream.close();
+}
 
 module.exports = {
+  destroyer,
   destroy,
   undestroy,
   errorOrDestroy
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index 9d38346742..1ead5cdf9f 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -12,6 +12,7 @@ const {
 let eos;
 
 const { once } = require('internal/util');
+const destroyImpl = require('internal/streams/destroy');
 const {
   ERR_INVALID_ARG_TYPE,
   ERR_INVALID_RETURN_VALUE,
@@ -24,10 +25,6 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
-function isRequest(stream) {
-  return stream && stream.setHeader && typeof stream.abort === 'function';
-}
-
 function destroyer(stream, reading, writing, callback) {
   callback = once(callback);
 
@@ -49,11 +46,7 @@ function destroyer(stream, reading, writing, callback) {
     if (destroyed) return;
     destroyed = true;
 
-    // request.destroy just do .end - .abort is what we want
-    if (isRequest(stream)) return stream.abort();
-    if (isRequest(stream.req)) return stream.req.abort();
-    if (typeof stream.destroy === 'function') return stream.destroy(err);
-    if (typeof stream.close === 'function') return stream.close();
+    destroyImpl.destroyer(stream, err);
 
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };