Skip to content

Commit 10a0718

Browse files
debadree25MattiasBuelens
debadree25
authored and
MattiasBuelens
committed
stream: implement TransformStream cleanup using "transformer.cancel"
Fixes: nodejs#49971 PR-URL: nodejs#50126 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent c530520 commit 10a0718

File tree

72 files changed

+418
-94
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

72 files changed

+418
-94
lines changed

‎lib/internal/webstreams/readablestream.js

+3-2
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ const {
1414
ObjectCreate,
1515
ObjectDefineProperties,
1616
ObjectSetPrototypeOf,
17+
Promise,
1718
PromisePrototypeThen,
1819
PromiseResolve,
1920
PromiseReject,
@@ -2441,7 +2442,7 @@ function setupReadableStreamDefaultController(
24412442
const startResult = startAlgorithm();
24422443

24432444
PromisePrototypeThen(
2444-
PromiseResolve(startResult),
2445+
new Promise((r) => r(startResult)),
24452446
() => {
24462447
controller[kState].started = true;
24472448
assert(!controller[kState].pulling);
@@ -3240,7 +3241,7 @@ function setupReadableByteStreamController(
32403241
const startResult = startAlgorithm();
32413242

32423243
PromisePrototypeThen(
3243-
PromiseResolve(startResult),
3244+
new Promise((r) => r(startResult)),
32443245
() => {
32453246
controller[kState].started = true;
32463247
assert(!controller[kState].pulling);

‎lib/internal/webstreams/transformstream.js

+105-11
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ const {
55
FunctionPrototypeCall,
66
ObjectDefineProperties,
77
PromisePrototypeThen,
8-
PromiseResolve,
98
ReflectConstruct,
109
SymbolToStringTag,
1110
Symbol,
@@ -47,6 +46,7 @@ const {
4746
nonOpFlush,
4847
kType,
4948
kState,
49+
nonOpCancel,
5050
} = require('internal/webstreams/util');
5151

5252
const {
@@ -377,8 +377,7 @@ function initializeTransformStream(
377377
return transformStreamDefaultSourcePullAlgorithm(stream);
378378
},
379379
cancel(reason) {
380-
transformStreamErrorWritableAndUnblockWrite(stream, reason);
381-
return PromiseResolve();
380+
return transformStreamDefaultSourceCancelAlgorithm(stream, reason);
382381
},
383382
}, {
384383
highWaterMark: readableHighWaterMark,
@@ -420,6 +419,10 @@ function transformStreamErrorWritableAndUnblockWrite(stream, error) {
420419
writableStreamDefaultControllerErrorIfNeeded(
421420
writable[kState].controller,
422421
error);
422+
transformStreamUnblockWrite(stream);
423+
}
424+
425+
function transformStreamUnblockWrite(stream) {
423426
if (stream[kState].backpressure)
424427
transformStreamSetBackpressure(stream, false);
425428
}
@@ -436,13 +439,15 @@ function setupTransformStreamDefaultController(
436439
stream,
437440
controller,
438441
transformAlgorithm,
439-
flushAlgorithm) {
442+
flushAlgorithm,
443+
cancelAlgorithm) {
440444
assert(isTransformStream(stream));
441445
assert(stream[kState].controller === undefined);
442446
controller[kState] = {
443447
stream,
444448
transformAlgorithm,
445449
flushAlgorithm,
450+
cancelAlgorithm,
446451
};
447452
stream[kState].controller = controller;
448453
}
@@ -453,21 +458,26 @@ function setupTransformStreamDefaultControllerFromTransformer(
453458
const controller = new TransformStreamDefaultController(kSkipThrow);
454459
const transform = transformer?.transform || defaultTransformAlgorithm;
455460
const flush = transformer?.flush || nonOpFlush;
461+
const cancel = transformer?.cancel || nonOpCancel;
456462
const transformAlgorithm =
457463
FunctionPrototypeBind(transform, transformer);
458464
const flushAlgorithm =
459465
FunctionPrototypeBind(flush, transformer);
466+
const cancelAlgorithm =
467+
FunctionPrototypeBind(cancel, transformer);
460468

461469
setupTransformStreamDefaultController(
462470
stream,
463471
controller,
464472
transformAlgorithm,
465-
flushAlgorithm);
473+
flushAlgorithm,
474+
cancelAlgorithm);
466475
}
467476

468477
function transformStreamDefaultControllerClearAlgorithms(controller) {
469478
controller[kState].transformAlgorithm = undefined;
470479
controller[kState].flushAlgorithm = undefined;
480+
controller[kState].cancelAlgorithm = undefined;
471481
}
472482

473483
function transformStreamDefaultControllerEnqueue(controller, chunk) {
@@ -556,7 +566,40 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
556566
}
557567

558568
async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
559-
transformStreamError(stream, reason);
569+
const {
570+
controller,
571+
readable,
572+
} = stream[kState];
573+
574+
if (controller[kState].finishPromise !== undefined) {
575+
return controller[kState].finishPromise;
576+
}
577+
578+
const { promise, resolve, reject } = createDeferredPromise();
579+
controller[kState].finishPromise = promise;
580+
const cancelPromise = ensureIsPromise(
581+
controller[kState].cancelAlgorithm,
582+
controller,
583+
reason);
584+
transformStreamDefaultControllerClearAlgorithms(controller);
585+
586+
PromisePrototypeThen(
587+
cancelPromise,
588+
() => {
589+
if (readable[kState].state === 'errored')
590+
reject(readable[kState].storedError);
591+
else {
592+
readableStreamDefaultControllerError(readable[kState].controller, reason);
593+
resolve();
594+
}
595+
},
596+
(error) => {
597+
readableStreamDefaultControllerError(readable[kState].controller, error);
598+
reject(error);
599+
},
600+
);
601+
602+
return controller[kState].finishPromise;
560603
}
561604

562605
function transformStreamDefaultSinkCloseAlgorithm(stream) {
@@ -565,23 +608,32 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
565608
controller,
566609
} = stream[kState];
567610

611+
if (controller[kState].finishPromise !== undefined) {
612+
return controller[kState].finishPromise;
613+
}
614+
const { promise, resolve, reject } = createDeferredPromise();
615+
controller[kState].finishPromise = promise;
568616
const flushPromise =
569617
ensureIsPromise(
570618
controller[kState].flushAlgorithm,
571619
controller,
572620
controller);
573621
transformStreamDefaultControllerClearAlgorithms(controller);
574-
return PromisePrototypeThen(
622+
PromisePrototypeThen(
575623
flushPromise,
576624
() => {
577625
if (readable[kState].state === 'errored')
578-
throw readable[kState].storedError;
579-
readableStreamDefaultControllerClose(readable[kState].controller);
626+
reject(readable[kState].storedError);
627+
else {
628+
readableStreamDefaultControllerClose(readable[kState].controller);
629+
resolve();
630+
}
580631
},
581632
(error) => {
582-
transformStreamError(stream, error);
583-
throw readable[kState].storedError;
633+
readableStreamDefaultControllerError(readable[kState].controller, error);
634+
reject(error);
584635
});
636+
return controller[kState].finishPromise;
585637
}
586638

587639
function transformStreamDefaultSourcePullAlgorithm(stream) {
@@ -591,6 +643,48 @@ function transformStreamDefaultSourcePullAlgorithm(stream) {
591643
return stream[kState].backpressureChange.promise;
592644
}
593645

646+
function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
647+
const {
648+
controller,
649+
writable,
650+
} = stream[kState];
651+
652+
if (controller[kState].finishPromise !== undefined) {
653+
return controller[kState].finishPromise;
654+
}
655+
656+
const { promise, resolve, reject } = createDeferredPromise();
657+
controller[kState].finishPromise = promise;
658+
const cancelPromise = ensureIsPromise(
659+
controller[kState].cancelAlgorithm,
660+
controller,
661+
reason);
662+
transformStreamDefaultControllerClearAlgorithms(controller);
663+
664+
PromisePrototypeThen(cancelPromise,
665+
() => {
666+
if (writable[kState].state === 'errored')
667+
reject(writable[kState].storedError);
668+
else {
669+
writableStreamDefaultControllerErrorIfNeeded(
670+
writable[kState].controller,
671+
reason);
672+
transformStreamUnblockWrite(stream);
673+
resolve();
674+
}
675+
},
676+
(error) => {
677+
writableStreamDefaultControllerErrorIfNeeded(
678+
writable[kState].controller,
679+
error);
680+
transformStreamUnblockWrite(stream);
681+
reject(error);
682+
},
683+
);
684+
685+
return controller[kState].finishPromise;
686+
}
687+
594688
module.exports = {
595689
TransformStream,
596690
TransformStreamDefaultController,

‎lib/internal/webstreams/writablestream.js

+2-1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ const {
66
FunctionPrototypeBind,
77
FunctionPrototypeCall,
88
ObjectDefineProperties,
9+
Promise,
910
PromisePrototypeThen,
1011
PromiseResolve,
1112
PromiseReject,
@@ -1290,7 +1291,7 @@ function setupWritableStreamDefaultController(
12901291
const startResult = startAlgorithm();
12911292

12921293
PromisePrototypeThen(
1293-
PromiseResolve(startResult),
1294+
new Promise((r) => r(startResult)),
12941295
() => {
12951296
assert(stream[kState].state === 'writable' ||
12961297
stream[kState].state === 'erroring');

‎test/fixtures/wpt/README.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ Last update:
2626
- performance-timeline: https://github.com/web-platform-tests/wpt/tree/17ebc3aea0/performance-timeline
2727
- resource-timing: https://github.com/web-platform-tests/wpt/tree/22d38586d0/resource-timing
2828
- resources: https://github.com/web-platform-tests/wpt/tree/919874f84f/resources
29-
- streams: https://github.com/web-platform-tests/wpt/tree/517e945bbf/streams
29+
- streams: https://github.com/web-platform-tests/wpt/tree/a8872d92b1/streams
3030
- url: https://github.com/web-platform-tests/wpt/tree/c2d7e70b52/url
3131
- user-timing: https://github.com/web-platform-tests/wpt/tree/5ae85bf826/user-timing
3232
- wasm/jsapi: https://github.com/web-platform-tests/wpt/tree/cde25e7e3c/wasm/jsapi

‎test/fixtures/wpt/streams/piping/abort.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/recording-streams.js
33
// META: script=../resources/test-utils.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/close-propagation-backward.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/recording-streams.js
33
'use strict';
44

‎test/fixtures/wpt/streams/piping/close-propagation-forward.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/crashtests/cross-piping.html

+12
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
<!DOCTYPE html>
2+
<script type="module">
3+
let a = new ReadableStream();
4+
let b = self.open()
5+
let f = new b.WritableStream();
6+
a.pipeThrough(
7+
{ "readable": a, "writable": f },
8+
{ "signal": AbortSignal.abort() }
9+
)
10+
await new Promise(setTimeout);
11+
structuredClone(undefined, { "transfer": [f] })
12+
</script>

‎test/fixtures/wpt/streams/piping/error-propagation-backward.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/error-propagation-forward.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/flow-control.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/rs-utils.js
44
// META: script=../resources/recording-streams.js

‎test/fixtures/wpt/streams/piping/general-addition.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
promise_test(async t => {

‎test/fixtures/wpt/streams/piping/general.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/multiple-propagation.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/pipe-through.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/rs-utils.js
33
// META: script=../resources/test-utils.js
44
// META: script=../resources/recording-streams.js

‎test/fixtures/wpt/streams/piping/then-interception.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/test-utils.js
33
// META: script=../resources/recording-streams.js
44
'use strict';

‎test/fixtures/wpt/streams/piping/throwing-options.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
class ThrowingOptions {

‎test/fixtures/wpt/streams/piping/transform-streams.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
promise_test(() => {

‎test/fixtures/wpt/streams/queuing-strategies.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
const highWaterMarkConversions = new Map([

‎test/fixtures/wpt/streams/readable-byte-streams/bad-buffers-and-views.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
promise_test(() => {

‎test/fixtures/wpt/streams/readable-byte-streams/construct-byob-request.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/rs-utils.js
33
'use strict';
44

‎test/fixtures/wpt/streams/readable-byte-streams/enqueue-with-detached-buffer.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22

33
promise_test(async t => {
44
const error = new Error('cannot proceed');

‎test/fixtures/wpt/streams/readable-byte-streams/general.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
// META: script=../resources/rs-utils.js
33
// META: script=../resources/test-utils.js
44
'use strict';

‎test/fixtures/wpt/streams/readable-byte-streams/non-transferable-buffers.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22
'use strict';
33

44
promise_test(async t => {

‎test/fixtures/wpt/streams/readable-byte-streams/respond-after-enqueue.any.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// META: global=window,worker
1+
// META: global=window,worker,shadowrealm
22

33
'use strict';
44

0 commit comments

Comments
 (0)