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

Commit 6d8e643

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 a082916 commit 6d8e643

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
@@ -409,12 +409,12 @@ class PushQueue {
409409
return this.#writerState;
410410
}
411411

412-
get pendingEndPromise() {
413-
return this.#pendingEnd?.promise ?? null;
414-
}
415-
416-
setPendingEnd(pending) {
417-
this.#pendingEnd = pending;
412+
getPendingEndPromise() {
413+
if (this.#pendingEnd === null) {
414+
const { promise, resolve, reject } = PromiseWithResolvers();
415+
this.#pendingEnd = { __proto__: null, promise, resolve, reject };
416+
}
417+
return this.#pendingEnd.promise;
418418
}
419419

420420
/**
@@ -681,15 +681,7 @@ class PushWriter {
681681
return PromiseReject(this.#queue.error);
682682
}
683683
if (result === -3) {
684-
// Closing: buffer has data, create deferred promise that resolves
685-
// when consumer drains past the end sentinel
686-
const pendingEndPromise = this.#queue.pendingEndPromise;
687-
if (pendingEndPromise !== null) {
688-
return raceEndWithSignal(pendingEndPromise, signal);
689-
}
690-
const { promise, resolve, reject } = PromiseWithResolvers();
691-
this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
692-
return raceEndWithSignal(promise, signal);
684+
return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal);
693685
}
694686
// >= 0: byte count (immediate close or idempotent)
695687
return PromiseResolve(result);
@@ -709,8 +701,7 @@ class PushWriter {
709701
[SymbolAsyncDispose]() {
710702
const state = this.#queue.writerState;
711703
if (state === 'closing') {
712-
// Wait for graceful drain
713-
return this.#queue.pendingEndPromise ?? PromiseResolve();
704+
return this.#queue.getPendingEndPromise();
714705
}
715706
if (state === 'open') {
716707
this.fail();

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

Lines changed: 18 additions & 0 deletions

0 commit comments

Comments
 (0)