File tree Expand file tree Collapse file tree
lib/internal/streams/iter Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -128,7 +128,7 @@ class BroadcastImpl {
128128 // own internal AbortController that follows the external signal.
129129 // When no transforms, return rawConsumer directly (controller elided
130130 // per PULL-02 optimization -- no transforms means no signal recipient).
131- if ( transforms . length > 0 ) {
131+ if ( transforms . length > 0 || options ?. signal ) {
132132 const pullArgs = [ ...transforms ] ;
133133 if ( options ?. signal ) {
134134 ArrayPrototypePush ( pullArgs ,
Original file line number Diff line number Diff line change @@ -97,7 +97,7 @@ class ShareImpl {
9797 const { transforms, options } = parsePullArgs ( args ) ;
9898 const rawConsumer = this . #createRawConsumer( ) ;
9999
100- if ( transforms . length > 0 ) {
100+ if ( transforms . length > 0 || options ?. signal ) {
101101 if ( options ) {
102102 return pullWithTransforms ( rawConsumer , ...transforms , options ) ;
103103 }
Original file line number Diff line number Diff line change @@ -174,6 +174,19 @@ async function testPendingNextSettlesAfterReturn() {
174174 assert . strictEqual ( result . value , undefined ) ;
175175}
176176
177+ async function testPushAbortSignalRejectsPendingNext ( ) {
178+ const ac = new AbortController ( ) ;
179+ const reason = new Error ( 'push aborted' ) ;
180+ const { broadcast : bc } = broadcast ( ) ;
181+ const iter = bc . push ( { signal : ac . signal } ) [ Symbol . asyncIterator ] ( ) ;
182+
183+ const pendingNext = iter . next ( ) ;
184+ const rejected = assert . rejects ( pendingNext , ( error ) => error === reason ) ;
185+ ac . abort ( reason ) ;
186+
187+ await rejected ;
188+ }
189+
177190// =============================================================================
178191// Writer fail detaches consumers
179192// =============================================================================
@@ -300,6 +313,7 @@ Promise.all([
300313 testCancelWithReason ( ) ,
301314 testCancelWithFalsyReason ( ) ,
302315 testPendingNextSettlesAfterReturn ( ) ,
316+ testPushAbortSignalRejectsPendingNext ( ) ,
303317 testFailDetachesConsumers ( ) ,
304318 testWriterFailIdempotent ( ) ,
305319 testLateJoinerSeesBufferedData ( ) ,
Original file line number Diff line number Diff line change @@ -196,6 +196,25 @@ async function testShareAbortSignalWhileSourcePullPending() {
196196 await Promise . all ( [ rejected1 , rejected2 ] ) ;
197197}
198198
199+ async function testSharePullAbortSignalRejectsPendingNext ( ) {
200+ const ac = new AbortController ( ) ;
201+ const reason = new Error ( 'pull aborted' ) ;
202+ const shared = share (
203+ // eslint-disable-next-line require-yield
204+ ( async function * never ( ) {
205+ await new Promise ( ( ) => { } ) ;
206+ } ) ( ) ,
207+ ) ;
208+ const iter = shared . pull ( { signal : ac . signal } ) [ Symbol . asyncIterator ] ( ) ;
209+
210+ const pendingNext = iter . next ( ) ;
211+ const rejected = assert . rejects ( pendingNext , ( error ) => error === reason ) ;
212+ ac . abort ( reason ) ;
213+
214+ await rejected ;
215+ shared . cancel ( ) ;
216+ }
217+
199218async function testShareAlreadyAborted ( ) {
200219 const ac = new AbortController ( ) ;
201220 ac . abort ( ) ;
@@ -343,6 +362,7 @@ Promise.all([
343362 testShareCancelWithReason ( ) ,
344363 testShareAbortSignal ( ) ,
345364 testShareAbortSignalWhileSourcePullPending ( ) ,
365+ testSharePullAbortSignalRejectsPendingNext ( ) ,
346366 testShareAlreadyAborted ( ) ,
347367 testShareSourceError ( ) ,
348368 testShareLateJoiningConsumer ( ) ,
You can’t perform that action at this time.
0 commit comments