stream: use addAbortListener

PR-URL: #48550 Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Matteo Collina <[email protected]>

Author: atlowChemi

lib/internal/streams/add-abort-signal.js

@@ -1,5 +1,9 @@
 'use strict';
 
+const{
+ SymbolDispose,
+}=primordials;
+
 const{
  AbortError,
  codes,
@@ -13,6 +17,7 @@ const{
 
 consteos=require('internal/streams/end-of-stream');
 const{ERR_INVALID_ARG_TYPE}=codes;
+letaddAbortListener;
 
 // This method is inlined here for readable-stream
 // It also does not allow for signal to not exist on the stream
@@ -46,8 +51,9 @@ module.exports.addAbortSignalNoValidate = function(signal, stream){
 if(signal.aborted){
 onAbort();
 }else{
-signal.addEventListener('abort',onAbort);
-eos(stream,()=>signal.removeEventListener('abort',onAbort));
+addAbortListener??=require('events').addAbortListener;
+constdisposable=addAbortListener(signal,onAbort);
+eos(stream,disposable[SymbolDispose]);
 }
 returnstream;
 };

lib/internal/streams/end-of-stream.js

@@ -22,7 +22,11 @@ const{
  validateBoolean,
 }=require('internal/validators');
 
-const{ Promise, PromisePrototypeThen }=primordials;
+const{
+ Promise,
+ PromisePrototypeThen,
+ SymbolDispose,
+}=primordials;
 
 const{
  isClosed,
@@ -40,6 +44,7 @@ const{
 willEmitClose: _willEmitClose,
  kIsClosedPromise,
 }=require('internal/streams/utils');
+letaddAbortListener;
 
 functionisRequest(stream){
 returnstream.setHeader&&typeofstream.abort==='function';
@@ -249,12 +254,13 @@ function eos(stream, options, callback){
 if(options.signal.aborted){
 process.nextTick(abort);
 }else{
+addAbortListener??=require('events').addAbortListener;
+constdisposable=addAbortListener(options.signal,abort);
 constoriginalCallback=callback;
 callback=once((...args)=>{
-options.signal.removeEventListener('abort',abort);
+disposable[SymbolDispose]();
 originalCallback.apply(stream,args);
 });
-options.signal.addEventListener('abort',abort);
 }
 }
 
@@ -272,12 +278,13 @@ function eosWeb(stream, options, callback){
 if(options.signal.aborted){
 process.nextTick(abort);
 }else{
+addAbortListener??=require('events').addAbortListener;
+constdisposable=addAbortListener(options.signal,abort);
 constoriginalCallback=callback;
 callback=once((...args)=>{
-options.signal.removeEventListener('abort',abort);
+disposable[SymbolDispose]();
 originalCallback.apply(stream,args);
 });
-options.signal.addEventListener('abort',abort);
 }
 }
 constresolverFn=(...args)=>{

lib/internal/streams/operators.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const{ AbortController }=require('internal/abort_controller');
+const{ AbortController, AbortSignal}=require('internal/abort_controller');
 
 const{
 codes: {
@@ -16,7 +16,7 @@ const{
  validateInteger,
  validateObject,
 }=require('internal/validators');
-const{ kWeakHandler }=require('internal/event_target');
+const{ kWeakHandler, kResistStopPropagation}=require('internal/event_target');
 const{ finished }=require('internal/streams/end-of-stream');
 conststaticCompose=require('internal/streams/compose');
 const{
@@ -27,6 +27,7 @@ const{deprecate } = require('internal/util');
 
 const{
  ArrayPrototypePush,
+ Boolean,
  MathFloor,
  Number,
  NumberIsNaN,
@@ -84,19 +85,11 @@ function map(fn, options){
 validateInteger(concurrency,'concurrency',1);
 
 returnasyncfunction*map(){
-constac=newAbortController();
+constsignal=AbortSignal.any([options?.signal].filter(Boolean));
 conststream=this;
 constqueue=[];
-constsignal=ac.signal;
 constsignalOpt={ signal };
 
-constabort=()=>ac.abort();
-if(options?.signal?.aborted){
-abort();
-}
-
-options?.signal?.addEventListener('abort',abort);
-
 letnext;
 letresume;
 letdone=false;
@@ -153,7 +146,6 @@ function map(fn, options){
 next();
 next=null;
 }
-options?.signal?.removeEventListener('abort',abort);
 }
 }
 
@@ -188,8 +180,6 @@ function map(fn, options){
 });
 }
 }finally{
-ac.abort();
-
 done=true;
 if(resume){
 resume();
@@ -301,7 +291,7 @@ async function reduce(reducer, initialValue, options){
 constac=newAbortController();
 constsignal=ac.signal;
 if(options?.signal){
-constopts={once: true,[kWeakHandler]: this};
+constopts={once: true,[kWeakHandler]: this,[kResistStopPropagation]: true};
 options.signal.addEventListener('abort',()=>ac.abort(),opts);
 }
 letgotAnyItemFromStream=false;

lib/internal/streams/pipeline.js

@@ -7,6 +7,7 @@ const{
  ArrayIsArray,
  Promise,
  SymbolAsyncIterator,
+ SymbolDispose,
 }=primordials;
 
 consteos=require('internal/streams/end-of-stream');
@@ -44,6 +45,7 @@ const{AbortController } = require('internal/abort_controller');
 
 letPassThrough;
 letReadable;
+letaddAbortListener;
 
 functiondestroyer(stream,reading,writing){
 letfinished=false;
@@ -206,7 +208,11 @@ function pipelineImpl(streams, callback, opts){
 finishImpl(newAbortError());
 }
 
-outerSignal?.addEventListener('abort',abort);
+addAbortListener??=require('events').addAbortListener;
+letdisposable;
+if(outerSignal){
+disposable=addAbortListener(outerSignal,abort);
+}
 
 leterror;
 letvalue;
@@ -231,7 +237,7 @@ function pipelineImpl(streams, callback, opts){
 destroys.shift()(error);
 }
 
-outerSignal?.removeEventListener('abort',abort);
+disposable?.[SymbolDispose]();
 ac.abort();
 
 if(final){

lib/internal/webstreams/readablestream.js

@@ -22,6 +22,7 @@ const{
  SafePromiseAll,
  Symbol,
  SymbolAsyncIterator,
+ SymbolDispose,
  SymbolToStringTag,
  Uint8Array,
 }=primordials;
@@ -140,6 +141,7 @@ const kRelease = Symbol('kRelease');
 
 letreleasedError;
 letreleasingError;
+letaddAbortListener;
 
 constuserModuleRegExp=/^{4}at(?:[^/\\(]+\()(?!node:(.+):\d+:\d+\)$).*/gm;
 
@@ -1259,6 +1261,7 @@ function readableStreamPipeTo(
 
 letreader;
 letwriter;
+letdisposable;
 // Both of these can throw synchronously. We want to capture
 // the error and return a rejected promise instead.
 try{
@@ -1291,7 +1294,7 @@ function readableStreamPipeTo(
 writableStreamDefaultWriterRelease(writer);
 readableStreamReaderGenericRelease(reader);
 if(signal!==undefined)
-signal.removeEventListener('abort',abortAlgorithm);
+disposable?.[SymbolDispose]();
 if(rejected)
 promise.reject(error);
 else
@@ -1418,7 +1421,8 @@ function readableStreamPipeTo(
 abortAlgorithm();
 returnpromise.promise;
 }
-signal.addEventListener('abort',abortAlgorithm,{once: true});
+addAbortListener??=require('events').addAbortListener;
+disposable=addAbortListener(signal,abortAlgorithm);
 }
 
 setPromiseHandled(run());