Skip to content

stream: more iter streams fixes #66030

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jasnell wants to merge 7 commits into from
Closed

stream: more iter streams fixes #66030

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
37bec3d
stream: ensure nested async iterators are cancellation aware
jasnell Aug 30, 2026
4acd6f1
stream: make broadcast waiters use SafeSet
jasnell Aug 30, 2026
00f4652
stream: ensure protocol and transform accessors are called once
jasnell Aug 30, 2026
8cc140e
stream: have broadcast clean factory listeners on cancel
jasnell Aug 30, 2026
efc2f37
stream: ensure streamable protocols precede array/validated-source fa…
jasnell Aug 30, 2026
a409c52
stream: make Broadcast.from() only return the Broadcast
jasnell Aug 30, 2026
d0ff99c
stream: update ShareImpl to retain factory abort listener
jasnell Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
stream: update ShareImpl to retain factory abort listener 
Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
  • Loading branch information
@jasnell
jasnell committed Sep 17, 2026
commit d0ff99c6694e8ada0940ed525ed53fa6821053e0
10 changes: 8 additions & 2 deletions lib/internal/streams/iter/share.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,7 +432,9 @@ class ShareImpl {
this.#sourceError = error;
this.#sourceExhausted = true;
} finally {
if (this.#sourceExhausted) this.#cleanupFactorySignal();
if (this.#sourceExhausted && this.#consumers.size === 0) {
this.#cleanupFactorySignal();
}
this.#pulling = false;
for (let i = 0; i < this.#pullWaiters.length; i++) {
this.#pullWaiters[i]();
Expand Down Expand Up @@ -496,7 +498,11 @@ class ShareImpl {

#deleteConsumer(consumer) {
if (this.#consumers.delete(consumer)) {
return this.#deleteConsumerFromMin(consumer);
const shouldTrimBuffer = this.#deleteConsumerFromMin(consumer);
if (this.#sourceExhausted && this.#consumers.size === 0) {
this.#cleanupFactorySignal();
}
return shouldTrimBuffer;
}
return false;
}
Expand Down
17 changes: 17 additions & 0 deletions test/parallel/test-stream-iter-factory-signal.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,22 @@ async function testShareSignalLifetime() {
assert.strictEqual(abortListenerCount(controller.signal), 0);
}

async function testShareSignalLifetimeWithSlowConsumer() {
const controller = new AbortController();
const reason = new Error('aborted');
const shared = share('x', { signal: controller.signal });
const fast = shared.pull()[Symbol.asyncIterator]();
const slow = shared.pull()[Symbol.asyncIterator]();

assert.strictEqual((await fast.next()).done, false);
assert.strictEqual((await fast.next()).done, true);
assert.strictEqual(abortListenerCount(controller.signal), 1);

controller.abort(reason);
await assert.rejects(slow.next(), (error) => error === reason);
assert.strictEqual(abortListenerCount(controller.signal), 0);
}

async function testDuplexSignalLifetime() {
const controller = new AbortController();
const [channelA, channelB] = duplex({ signal: controller.signal });
Expand All @@ -80,5 +96,6 @@ Promise.all([
testPushSignalLifetime(),
testBroadcastSignalLifetime(),
testShareSignalLifetime(),
testShareSignalLifetimeWithSlowConsumer(),
testDuplexSignalLifetime(),
]).then(common.mustCall());
Loading