GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — stream: honor AbortSignal in Writer.end() · nodejs/node@420c3ef

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • lib/internal/streams/iter
    • broadcast.js
    • push.js
  • test/parallel
    • test-stream-iter-broadcast-basic.js
    • test-stream-iter-duplex.js
    • test-stream-iter-push-writer.js
Expand file treeCollapse file tree
Open diff view settings
Collapse file

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

Copy file name to clipboardExpand all lines: lib/internal/streams/iter/broadcast.js
+3-2Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -630,9 +630,10 @@ class BroadcastWriter {
630630
return false;
631631
}
632632

633-
// end() is synchronous internally - signal accepted for interface compliance.
634633
end(options) {
635-
getWriterSignal(options);
634+
const signal = getWriterSignal(options);
635+
if (signal?.aborted) return PromiseReject(signal.reason);
636+
636637
if (this.#isClosed()) return this.#closed;
637638
this.#closed = PromiseResolve(this.#totalBytes);
638639
this.#broadcast[kEnd]();
Collapse file

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

Copy file name to clipboardExpand all lines: lib/internal/streams/iter/push.js
+35-3Lines changed: 35 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
const {
99
ArrayIsArray,
1010
ArrayPrototypePush,
11+
PromisePrototypeThen,
1112
PromiseReject,
1213
PromiseResolve,
1314
PromiseWithResolvers,
@@ -55,6 +56,35 @@ const {
5556

5657
const kNoFailReason = Symbol('kNoFailReason');
5758

59+
function raceEndWithSignal(promise, signal) {
60+
if (!signal) return promise;
61+
if (signal.aborted) return PromiseReject(signal.reason);
62+
63+
const {
64+
promise: signaledPromise,
65+
resolve,
66+
reject,
67+
} = PromiseWithResolvers();
68+
const onAbort = () => reject(signal.reason);
69+
70+
signal.addEventListener('abort', onAbort, {
71+
__proto__: null,
72+
once: true,
73+
});
74+
PromisePrototypeThen(
75+
promise,
76+
(value) => {
77+
signal.removeEventListener('abort', onAbort);
78+
resolve(value);
79+
},
80+
(reason) => {
81+
signal.removeEventListener('abort', onAbort);
82+
reject(reason);
83+
},
84+
);
85+
return signaledPromise;
86+
}
87+
5888
// =============================================================================
5989
// PushQueue - Internal Queue with Chunk-Based Backpressure
6090
// =============================================================================
@@ -628,7 +658,9 @@ class PushWriter {
628658
}
629659

630660
end(options) {
631-
getWriterSignal(options);
661+
const signal = getWriterSignal(options);
662+
if (signal?.aborted) return PromiseReject(signal.reason);
663+
632664
const result = this.#queue.end();
633665
if (result === -2) {
634666
// Errored: reject with stored error
@@ -639,11 +671,11 @@ class PushWriter {
639671
// when consumer drains past the end sentinel
640672
const pendingEndPromise = this.#queue.pendingEndPromise;
641673
if (pendingEndPromise !== null) {
642-
return pendingEndPromise;
674+
return raceEndWithSignal(pendingEndPromise, signal);
643675
}
644676
const { promise, resolve, reject } = PromiseWithResolvers();
645677
this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject });
646-
return promise;
678+
return raceEndWithSignal(promise, signal);
647679
}
648680
// >= 0: byte count (immediate close or idempotent)
649681
return PromiseResolve(result);
Collapse file

‎test/parallel/test-stream-iter-broadcast-basic.js‎

Copy file name to clipboardExpand all lines: test/parallel/test-stream-iter-broadcast-basic.js
+17Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,22 @@ async function testWriterEnd() {
111111
assert.strictEqual(data, 'data');
112112
}
113113

114+
async function testWriterEndWithPreAbortedSignal() {
115+
const { writer, broadcast: bc } = broadcast();
116+
const consumer = bc.push();
117+
const reason = new Error('end aborted');
118+
119+
await assert.rejects(
120+
writer.end({ signal: AbortSignal.abort(reason) }),
121+
(error) => error === reason,
122+
);
123+
124+
// A rejected end must leave the writer open.
125+
await writer.write('data');
126+
assert.strictEqual(await writer.end(), 4);
127+
assert.strictEqual(await text(consumer), 'data');
128+
}
129+
114130
async function testWriterFail() {
115131
const { writer, broadcast: bc } = broadcast();
116132
const consumer = bc.push();
@@ -308,6 +324,7 @@ Promise.all([
308324
testWriteSync(),
309325
testWritevSync(),
310326
testWriterEnd(),
327+
testWriterEndWithPreAbortedSignal(),
311328
testWriterFail(),
312329
testCancelWithoutReason(),
313330
testCancelWithReason(),
Collapse file

‎test/parallel/test-stream-iter-duplex.js‎

Copy file name to clipboardExpand all lines: test/parallel/test-stream-iter-duplex.js
+17Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,22 @@ async function testAbortSignal() {
127127
);
128128
}
129129

130+
async function testWriterEndWithPreAbortedSignal() {
131+
const [channelA, channelB] = duplex();
132+
const reason = new Error('end aborted');
133+
134+
await assert.rejects(
135+
channelA.writer.end({ signal: AbortSignal.abort(reason) }),
136+
(error) => error === reason,
137+
);
138+
139+
await channelA.writer.write('still open');
140+
const completedEnd = channelA.writer.end();
141+
assert.strictEqual(await text(channelB.readable), 'still open');
142+
assert.strictEqual(await completedEnd, 10);
143+
await channelB.close();
144+
}
145+
130146
async function testEmptyDuplex() {
131147
const [channelA, channelB] = duplex();
132148

@@ -182,6 +198,7 @@ Promise.all([
182198
testWithOptions(),
183199
testPerChannelOptions(),
184200
testAbortSignal(),
201+
testWriterEndWithPreAbortedSignal(),
185202
testEmptyDuplex(),
186203
testChannelFail(),
187204
testAbortSignalBothChannels(),
Collapse file

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

Copy file name to clipboardExpand all lines: test/parallel/test-stream-iter-push-writer.js
+36Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,40 @@ async function testEndAsyncReturnValue() {
260260
await consume;
261261
}
262262

263+
async function testEndWithPreAbortedSignal() {
264+
const { writer, readable } = push();
265+
const reason = new Error('end aborted');
266+
267+
writer.writeSync('hello');
268+
await assert.rejects(
269+
writer.end({ signal: AbortSignal.abort(reason) }),
270+
(error) => error === reason,
271+
);
272+
273+
// A rejected end must leave the writer open.
274+
writer.writeSync(' world');
275+
const consume = text(readable);
276+
assert.strictEqual(await writer.end(), 11);
277+
assert.strictEqual(await consume, 'hello world');
278+
}
279+
280+
async function testEndSignalAbortWhileDraining() {
281+
const { writer, readable } = push();
282+
const controller = new AbortController();
283+
const reason = new Error('end aborted while draining');
284+
285+
writer.writeSync('hello');
286+
const abortedEnd = writer.end({ signal: controller.signal });
287+
controller.abort(reason);
288+
289+
await assert.rejects(abortedEnd, (error) => error === reason);
290+
291+
// Aborting the operation does not undo the end-of-stream signal.
292+
const completedEnd = writer.end();
293+
assert.strictEqual(await text(readable), 'hello');
294+
assert.strictEqual(await completedEnd, 5);
295+
}
296+
263297
async function testEndAfterEndSyncWaitsForDrain() {
264298
const { writer, readable } = push();
265299
writer.writeSync('hello');
@@ -553,6 +587,8 @@ Promise.all([
553587
testOndrainProtocolErrorPropagates(),
554588
testFail(),
555589
testEndAsyncReturnValue(),
590+
testEndWithPreAbortedSignal(),
591+
testEndSignalAbortWhileDraining(),
556592
testEndAfterEndSyncWaitsForDrain(),
557593
testWriteUint8Array(),
558594
testOndrainWaitsForDrain(),

0 commit comments

Comments
0 (0)
You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.