stream: re-use legacy destroyer

ronag · MylesBorins · commit 4b04bf89ad1d · 2020-03-10T17:53:03.000-04:00
Backport-PR-URL: #32174 PR-URL: #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>
diff --git 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);
 });
 }
 
@@ -172,7 +164,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
@@ -128,8 +128,21 @@ function errorOrDestroy(stream, err) {
 stream.emit('error', err);
 }
 
+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
@@ -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,
@@ -28,14 +29,6 @@ function isRequest(stream) {
 return stream && stream.setHeader && typeof stream.abort === 'function';
 }
 
-function destroyStream(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();
-}
-
 function destroyer(stream, reading, writing, final, callback) {
 callback = once(callback);
 let destroyed = false;
@@ -46,15 +39,15 @@ function destroyer(stream, reading, writing, final, callback) {
 destroyed = true;
 const readable = stream.readable || isRequest(stream);
 if (err || !final || !readable) {
- destroyStream(stream, err);
+ destroyImpl.destroyer(stream, err);
 }
 callback(err);
 });
 
 return (err) => {
 if (destroyed) return;
 destroyed = true;
- destroyStream(stream, err);
+ destroyImpl.destroyer(stream, err);
 callback(err || new ERR_STREAM_DESTROYED('pipe'));
 };
 }

-Original file line number
+Diff line change
 }=primordials;
 constfinished=require('internal/streams/end-of-stream');
 +constdestroyImpl=require('internal/streams/destroy');
 constkLastResolve=Symbol('lastResolve');
 constkLastReject=Symbol('lastReject');
 letReadable;
 -functiondestroy(stream,err){
 -// request.destroy just do .end - .abort is what we want
 -if(typeofstream.abort==='function')returnstream.abort();
 -if(stream.req&&
 -typeofstream.req.abort==='function')returnstream.req.abort();
 -if(typeofstream.destroy==='function')returnstream.destroy(err);
 -if(typeofstream.close==='function')returnstream.close();
 -}
+-
 functioncreateIterResult(value,done){
 return{ value, done };
+}
 resolve(createIterResult(undefined,true));
+}
 });
 -destroy(stream,err);
 +destroyImpl.destroyer(stream,err);
 });
+}
 constsrc=stream;
 stream=newReadable({objectMode: true}).wrap(src);
 -finished(stream,(err)=>destroy(src,err));
 +finished(stream,(err)=>destroyImpl.destroyer(src,err));
+}
 constiterator=ObjectCreate(ReadableStreamAsyncIteratorPrototype,{
-Original file line number
+Diff line change
 leteos;
 const{ once }=require('internal/util');
 +constdestroyImpl=require('internal/streams/destroy');
 const{
 ERR_INVALID_ARG_TYPE,
 ERR_INVALID_RETURN_VALUE,
 returnstream&&stream.setHeader&&typeofstream.abort==='function';
+}
 -functiondestroyStream(stream,err){
 -// request.destroy just do .end - .abort is what we want
 -if(isRequest(stream))returnstream.abort();
 -if(isRequest(stream.req))returnstream.req.abort();
 -if(typeofstream.destroy==='function')returnstream.destroy(err);
 -if(typeofstream.close==='function')returnstream.close();
 -}
+-
 functiondestroyer(stream,reading,writing,final,callback){
 callback=once(callback);
 letdestroyed=false;
 destroyed=true;
 constreadable=stream.readable||isRequest(stream);
 if(err||!final||!readable){
 -destroyStream(stream,err);
 +destroyImpl.destroyer(stream,err);
+}
 callback(err);
 });
 return(err)=>{
 if(destroyed)return;
 destroyed=true;
 -destroyStream(stream,err);
 +destroyImpl.destroyer(stream,err);
 callback(err||newERR_STREAM_DESTROYED('pipe'));
 };
+}