stream: ensure iterator cleanup on done, reject, etc · nodejs/node@2130738 · GitHub
Skip to content

Commit 2130738

Browse files
jasnelladuh95
jasnell
authored and
aduh95
committed
stream: ensure iterator cleanup on done, reject, etc
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode PR-URL: #65652 Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 748adc6 commit 2130738

5 files changed

Lines changed: 115 additions & 23 deletions

File tree

‎lib/internal/streams/iter/broadcast.js‎

Lines changed: 2 additions & 7 deletions

‎lib/internal/streams/iter/pull.js‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ const {
1212
ArrayIsArray,
1313
ArrayPrototypePush,
1414
ArrayPrototypeSlice,
15+
FunctionPrototypeCall,
1516
PromisePrototypeThen,
1617
PromiseResolve,
1718
SymbolAsyncIterator,
@@ -896,6 +897,76 @@ function pull(source, ...args) {
896897
};
897898
}
898899

900+
// Keep ownership of a bonded consumer outside the transform pipeline so it can
901+
// be detached even when the pipeline never starts or terminates early.
902+
function pullWithConsumerCleanup(source, transforms, signal) {
903+
const sourceIterator = source[SymbolAsyncIterator]();
904+
const pipelineSource = {
905+
__proto__: null,
906+
[SymbolAsyncIterator]() {
907+
return sourceIterator;
908+
},
909+
};
910+
const pipeline = signal === undefined ?
911+
pull(pipelineSource, ...transforms) :
912+
pull(pipelineSource, ...transforms, { __proto__: null, signal });
913+
let sourceClosed = false;
914+
let abortHandler;
915+
916+
function closeSource(method, value) {
917+
if (sourceClosed) return;
918+
sourceClosed = true;
919+
if (abortHandler !== undefined) {
920+
signal.removeEventListener('abort', abortHandler);
921+
}
922+
const close = sourceIterator[method] ?? sourceIterator.return;
923+
if (typeof close === 'function') {
924+
const result = FunctionPrototypeCall(close, sourceIterator, value);
925+
PromisePrototypeThen(PromiseResolve(result), undefined, () => {});
926+
}
927+
}
928+
929+
if (signal !== undefined) {
930+
abortHandler = () => closeSource('throw', signal.reason);
931+
signal.addEventListener('abort', abortHandler,
932+
{ __proto__: null, once: true });
933+
if (signal.aborted) abortHandler();
934+
}
935+
936+
return {
937+
__proto__: null,
938+
[SymbolAsyncIterator]() {
939+
const iterator = pipeline[SymbolAsyncIterator]();
940+
return {
941+
__proto__: null,
942+
next(value) {
943+
return PromisePrototypeThen(
944+
iterator.next(value),
945+
(result) => {
946+
if (result.done) closeSource('return');
947+
return result;
948+
},
949+
(error) => {
950+
closeSource('throw', error);
951+
throw error;
952+
});
953+
},
954+
return(value) {
955+
closeSource('return', value);
956+
return iterator.return(value);
957+
},
958+
throw(error) {
959+
closeSource('throw', error);
960+
return iterator.throw(error);
961+
},
962+
[SymbolAsyncIterator]() {
963+
return this;
964+
},
965+
};
966+
},
967+
};
968+
}
969+
899970
// =============================================================================
900971
// Public API: pipeTo() and pipeToSync()
901972
// =============================================================================
@@ -1112,4 +1183,5 @@ module.exports = {
11121183
pipeToSync,
11131184
pull,
11141185
pullSync,
1186+
pullWithConsumerCleanup,
11151187
};

‎lib/internal/streams/iter/push.js‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ const {
4646
} = require('internal/streams/iter/utils');
4747

4848
const {
49-
pull: pullWithTransforms,
49+
pullWithConsumerCleanup,
5050
} = require('internal/streams/iter/pull');
5151

5252
const {
@@ -756,12 +756,8 @@ function push(...args) {
756756
// Apply transforms lazily if provided
757757
let readable;
758758
if (transforms.length > 0) {
759-
if (options.signal) {
760-
readable = pullWithTransforms(
761-
rawReadable, ...transforms, { __proto__: null, signal: options.signal });
762-
} else {
763-
readable = pullWithTransforms(rawReadable, ...transforms);
764-
}
759+
readable = pullWithConsumerCleanup(
760+
rawReadable, transforms, options.signal);
765761
} else {
766762
readable = rawReadable;
767763
}

‎lib/internal/streams/iter/share.js‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@ const {
3030
} = require('internal/streams/iter/from');
3131

3232
const {
33-
pull: pullWithTransforms,
3433
pullSync: pullSyncWithTransforms,
34+
pullWithConsumerCleanup,
3535
} = require('internal/streams/iter/pull');
3636

3737
const {
@@ -111,13 +111,7 @@ class ShareImpl {
111111
const rawConsumer = this.#createRawConsumer();
112112

113113
if (transforms.length > 0 || signal) {
114-
if (signal) {
115-
return pullWithTransforms(
116-
rawConsumer,
117-
...transforms,
118-
{ __proto__: null, signal });
119-
}
120-
return pullWithTransforms(rawConsumer, ...transforms);
114+
return pullWithConsumerCleanup(rawConsumer, transforms, signal);
121115
}
122116
return rawConsumer;
123117
}

‎test/parallel/test-stream-iter-pull-async.js‎

Lines changed: 36 additions & 1 deletion

0 commit comments

Comments
 (0)