stream: ensure async dispoal after endSync awaits for drain · nodejs/node@7567734 · GitHub
Skip to content

Commit 7567734

Browse files
jasnelladuh95
authored andcommitted
stream: ensure async dispoal after endSync awaits for drain
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode PR-URL: #65658 Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
1 parent 042b951 commit 7567734

3 files changed

Lines changed: 32 additions & 17 deletions

File tree

‎doc/api/stream_iter.md‎

Lines changed: 6 additions & 0 deletions

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

Lines changed: 8 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -412,12 +412,12 @@ class PushQueue {
412412
return this.#writerState;
413413
}
414414

415-
get pendingEndPromise() {
416-
return this.#pendingEnd?.promise ?? null;
417-
}
418-
419-
setPendingEnd(pending) {
420-
this.#pendingEnd = pending;
415+
getPendingEndPromise() {
416+
if (this.#pendingEnd === null) {
417+
const { promise, resolve, reject } = PromiseWithResolvers();
418+
this.#pendingEnd = { __proto__: null, promise, resolve, reject };
419+
}
420+
return this.#pendingEnd.promise;
421421
}
422422

423423
/**
@@ -691,15 +691,7 @@ class PushWriter {
691691
return PromiseReject(this.#queue.error);
692692
}
693693
if (result === -3) {
694-
// Closing: buffer has data, create deferred promise that resolves
695-
// when consumer drains past the end sentinel
696-
const pendingEndPromise = this.#queue.pendingEndPromise;
697-
if (pendingEndPromise !== null) {
698-
return raceEndWithSignal(pendingEndPromise, signal);
699-
}
700-
const { promise, resolve, reject } = PromiseWithResolvers();
701-
this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
702-
return raceEndWithSignal(promise, signal);
694+
return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal);
703695
}
704696
// >= 0: byte count (immediate close or idempotent)
705697
return PromiseResolve(result);
@@ -719,8 +711,7 @@ class PushWriter {
719711
[SymbolAsyncDispose]() {
720712
const state = this.#queue.writerState;
721713
if (state === 'closing') {
722-
// Wait for graceful drain
723-
return this.#queue.pendingEndPromise ?? PromiseResolve();
714+
return this.#queue.getPendingEndPromise();
724715
}
725716
if (state === 'open') {
726717
this.fail();

‎test/parallel/test-stream-iter-push-writer.js‎

Lines changed: 18 additions & 0 deletions

0 commit comments

Comments
 (0)