11'use strict' ;
22
3- const { AbortController } = require ( 'internal/abort_controller' ) ;
3+ const { AbortController, AbortSignal } = require ( 'internal/abort_controller' ) ;
44
55const {
66 codes : {
@@ -16,7 +16,7 @@ const {
1616 validateInteger,
1717 validateObject,
1818} = require ( 'internal/validators' ) ;
19- const { kWeakHandler } = require ( 'internal/event_target' ) ;
19+ const { kWeakHandler, kResistStopPropagation } = require ( 'internal/event_target' ) ;
2020const { finished } = require ( 'internal/streams/end-of-stream' ) ;
2121const staticCompose = require ( 'internal/streams/compose' ) ;
2222const {
@@ -27,6 +27,7 @@ const { deprecate } = require('internal/util');
2727
2828const {
2929 ArrayPrototypePush,
30+ Boolean,
3031 MathFloor,
3132 Number,
3233 NumberIsNaN,
@@ -84,19 +85,11 @@ function map(fn, options) {
8485 validateInteger ( concurrency , 'concurrency' , 1 ) ;
8586
8687 return async function * map ( ) {
87- const ac = new AbortController ( ) ;
88+ const signal = AbortSignal . any ( [ options ?. signal ] . filter ( Boolean ) ) ;
8889 const stream = this ;
8990 const queue = [ ] ;
90- const signal = ac . signal ;
9191 const signalOpt = { signal } ;
9292
93- const abort = ( ) => ac . abort ( ) ;
94- if ( options ?. signal ?. aborted ) {
95- abort ( ) ;
96- }
97-
98- options ?. signal ?. addEventListener ( 'abort' , abort ) ;
99-
10093 let next ;
10194 let resume ;
10295 let done = false ;
@@ -153,7 +146,6 @@ function map(fn, options) {
153146 next ( ) ;
154147 next = null ;
155148 }
156- options ?. signal ?. removeEventListener ( 'abort' , abort ) ;
157149 }
158150 }
159151
@@ -188,8 +180,6 @@ function map(fn, options) {
188180 } ) ;
189181 }
190182 } finally {
191- ac . abort ( ) ;
192-
193183 done = true ;
194184 if ( resume ) {
195185 resume ( ) ;
@@ -301,7 +291,7 @@ async function reduce(reducer, initialValue, options) {
301291 const ac = new AbortController ( ) ;
302292 const signal = ac . signal ;
303293 if ( options ?. signal ) {
304- const opts = { once : true , [ kWeakHandler ] : this } ;
294+ const opts = { once : true , [ kWeakHandler ] : this , [ kResistStopPropagation ] : true } ;
305295 options . signal . addEventListener ( 'abort' , ( ) => ac . abort ( ) , opts ) ;
306296 }
307297 let gotAnyItemFromStream = false ;
0 commit comments