stream: have broadcast clean factory listeners on cancel · nodejs/node@cdeb53e · GitHub
Skip to content

Commit cdeb53e

Browse files
jasnelladuh95
authored andcommitted
stream: have broadcast clean factory listeners on cancel
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode PR-URL: #66030 Reviewed-By: Filip Skokan <panva.ip@gmail.com> Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
1 parent 84f9031 commit cdeb53e

4 files changed

Lines changed: 145 additions & 25 deletions

File tree

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

Lines changed: 18 additions & 5 deletions

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

Lines changed: 26 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,10 @@ function duplex(options = { __proto__: null }) {
4545
backpressure: b?.backpressure ?? backpressure,
4646
});
4747

48-
const channelA = createDuplexChannel(aWriter, aReadable);
49-
const channelB = createDuplexChannel(bWriter, bReadable);
48+
let cleanupSignal;
49+
const onClose = () => cleanupSignal?.();
50+
const channelA = createDuplexChannel(aWriter, aReadable, onClose);
51+
const channelB = createDuplexChannel(bWriter, bReadable, onClose);
5052

5153
// Signal handler: fail both writers with the abort reason so consumers
5254
// see the error. This is an error-path shutdown, not a clean close.
@@ -55,6 +57,11 @@ function duplex(options = { __proto__: null }) {
5557
const reason = signal.reason;
5658
aWriter.fail(reason);
5759
bWriter.fail(reason);
60+
cleanupSignal?.();
61+
};
62+
cleanupSignal = () => {
63+
signal.removeEventListener('abort', abortBoth);
64+
cleanupSignal = undefined;
5865
};
5966
if (signal.aborted) {
6067
abortBoth();
@@ -67,7 +74,7 @@ function duplex(options = { __proto__: null }) {
6774
return [channelA, channelB];
6875
}
6976

70-
function createDuplexChannel(writer, readable) {
77+
function createDuplexChannel(writer, readable, onClose) {
7178
// A push readable has one shared consumer state. Keeping an iterator from
7279
// creation lets close() terminate that state even if no caller has iterated.
7380
const closeIterator = readable[SymbolAsyncIterator]();
@@ -78,7 +85,7 @@ function createDuplexChannel(writer, readable) {
7885
get writer() { return writer; },
7986
get readable() { return readable; },
8087
close() {
81-
closePromise ??= closeDuplexChannel(writer, closeIterator);
88+
closePromise ??= closeDuplexChannel(writer, closeIterator, onClose);
8289
return closePromise;
8390
},
8491
[SymbolAsyncDispose]() {
@@ -87,19 +94,23 @@ function createDuplexChannel(writer, readable) {
8794
};
8895
}
8996

90-
async function closeDuplexChannel(writer, closeIterator) {
91-
const result = writer.endSync();
92-
const endPromise = result < 0 ? writer.end() : undefined;
93-
const returnPromise = closeIterator.return();
97+
async function closeDuplexChannel(writer, closeIterator, onClose) {
98+
try {
99+
const result = writer.endSync();
100+
const endPromise = result < 0 ? writer.end() : undefined;
101+
const returnPromise = closeIterator.return();
94102

95-
if (endPromise !== undefined) {
96-
try {
97-
await SafePromiseAllReturnVoid([endPromise, returnPromise]);
98-
} catch (error) {
99-
if (!isConsumerReturnError(error)) throw error;
103+
if (endPromise !== undefined) {
104+
try {
105+
await SafePromiseAllReturnVoid([endPromise, returnPromise]);
106+
} catch (error) {
107+
if (!isConsumerReturnError(error)) throw error;
108+
}
109+
} else {
110+
await returnPromise;
100111
}
101-
} else {
102-
await returnPromise;
112+
} finally {
113+
onClose();
103114
}
104115
}
105116

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

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ const {
7171

7272
const kNoShareError = Symbol('kNoShareError');
7373
const kShareCancelled = Symbol('kShareCancelled');
74+
const kSetFactorySignal = Symbol('kSetFactorySignal');
7475

7576
class ShareImpl {
7677
#source;
@@ -89,6 +90,7 @@ class ShareImpl {
8990
#cancelError = kNoShareError;
9091
#cachedMinCursor = 0;
9192
#cachedMinCursorConsumers = 0;
93+
#abortHandler;
9294
/** Cumulative byte size of buffered entries */
9395
#bufferedBytes = 0;
9496

@@ -104,6 +106,11 @@ class ShareImpl {
104106
return this.#consumers.size;
105107
}
106108

109+
[kSetFactorySignal](signal) {
110+
this.#abortHandler = () => this.cancel(signal.reason);
111+
onSignalAbort(signal, this.#abortHandler);
112+
}
113+
107114
pull(...args) {
108115
const parsed = parsePullArgs(args);
109116
const { transforms } = parsed;
@@ -303,6 +310,7 @@ class ShareImpl {
303310
this.#consumers.clear();
304311
this.#buffer.clear();
305312
this.#bufferedBytes = 0;
313+
this.#cleanupFactorySignal();
306314

307315
for (let i = 0; i < this.#pullWaiters.length; i++) {
308316
this.#pullWaiters[i]();
@@ -424,6 +432,7 @@ class ShareImpl {
424432
this.#sourceError = error;
425433
this.#sourceExhausted = true;
426434
} finally {
435+
if (this.#sourceExhausted) this.#cleanupFactorySignal();
427436
this.#pulling = false;
428437
for (let i = 0; i < this.#pullWaiters.length; i++) {
429438
this.#pullWaiters[i]();
@@ -470,6 +479,13 @@ class ShareImpl {
470479
this.#cachedMinCursorConsumers = minCursorConsumers;
471480
}
472481

482+
#cleanupFactorySignal() {
483+
if (this.#abortHandler !== undefined) {
484+
this.#options.signal.removeEventListener('abort', this.#abortHandler);
485+
this.#abortHandler = undefined;
486+
}
487+
}
488+
473489
#deleteConsumerFromMin(consumer) {
474490
if (consumer.cursor === this.#cachedMinCursor) {
475491
this.#cachedMinCursorConsumers--;
@@ -768,10 +784,6 @@ class SyncShareImpl {
768784
}
769785
}
770786

771-
function onShareCancel(shareImpl, signal) {
772-
onSignalAbort(signal, () => shareImpl.cancel(signal.reason));
773-
}
774-
775787
// =============================================================================
776788
// Public API
777789
// =============================================================================
@@ -800,7 +812,7 @@ function share(source, options = { __proto__: null }) {
800812
const shareImpl = new ShareImpl(normalized, opts);
801813

802814
if (signal) {
803-
onShareCancel(shareImpl, signal);
815+
shareImpl[kSetFactorySignal](signal);
804816
}
805817

806818
return shareImpl;
Lines changed: 84 additions & 0 deletions

0 commit comments

Comments
 (0)