|
6 | 6 | ObjectDefineProperties, |
7 | 7 | ObjectSetPrototypeOf, |
8 | 8 | PromisePrototypeThen, |
9 | | - PromiseResolve, |
10 | 9 | SymbolToStringTag, |
11 | 10 | Symbol, |
12 | 11 | } = primordials; |
@@ -47,6 +46,7 @@ const { |
47 | 46 | nonOpFlush, |
48 | 47 | kType, |
49 | 48 | kState, |
| 49 | + nonOpCancel, |
50 | 50 | } = require('internal/webstreams/util'); |
51 | 51 |
|
52 | 52 | const { |
@@ -384,8 +384,7 @@ function initializeTransformStream( |
384 | 384 | return transformStreamDefaultSourcePullAlgorithm(stream); |
385 | 385 | }, |
386 | 386 | cancel(reason) { |
387 | | - transformStreamErrorWritableAndUnblockWrite(stream, reason); |
388 | | - return PromiseResolve(); |
| 387 | + return transformStreamDefaultSourceCancelAlgorithm(stream, reason); |
389 | 388 | }, |
390 | 389 | }, { |
391 | 390 | highWaterMark: readableHighWaterMark, |
@@ -427,6 +426,10 @@ function transformStreamErrorWritableAndUnblockWrite(stream, error) { |
427 | 426 | writableStreamDefaultControllerErrorIfNeeded( |
428 | 427 | writable[kState].controller, |
429 | 428 | error); |
| 429 | + transformStreamUnblockWrite(stream); |
| 430 | +} |
| 431 | + |
| 432 | +function transformStreamUnblockWrite(stream) { |
430 | 433 | if (stream[kState].backpressure) |
431 | 434 | transformStreamSetBackpressure(stream, false); |
432 | 435 | } |
@@ -443,13 +446,15 @@ function setupTransformStreamDefaultController( |
443 | 446 | stream, |
444 | 447 | controller, |
445 | 448 | transformAlgorithm, |
446 | | - flushAlgorithm) { |
| 449 | + flushAlgorithm, |
| 450 | + cancelAlgorithm) { |
447 | 451 | assert(isTransformStream(stream)); |
448 | 452 | assert(stream[kState].controller === undefined); |
449 | 453 | controller[kState] = { |
450 | 454 | stream, |
451 | 455 | transformAlgorithm, |
452 | 456 | flushAlgorithm, |
| 457 | + cancelAlgorithm, |
453 | 458 | }; |
454 | 459 | stream[kState].controller = controller; |
455 | 460 | } |
@@ -460,21 +465,26 @@ function setupTransformStreamDefaultControllerFromTransformer( |
460 | 465 | const controller = new TransformStreamDefaultController(kSkipThrow); |
461 | 466 | const transform = transformer?.transform || defaultTransformAlgorithm; |
462 | 467 | const flush = transformer?.flush || nonOpFlush; |
| 468 | + const cancel = transformer?.cancel || nonOpCancel; |
463 | 469 | const transformAlgorithm = |
464 | 470 | FunctionPrototypeBind(transform, transformer); |
465 | 471 | const flushAlgorithm = |
466 | 472 | FunctionPrototypeBind(flush, transformer); |
| 473 | + const cancelAlgorithm = |
| 474 | + FunctionPrototypeBind(cancel, transformer); |
467 | 475 |
|
468 | 476 | setupTransformStreamDefaultController( |
469 | 477 | stream, |
470 | 478 | controller, |
471 | 479 | transformAlgorithm, |
472 | | - flushAlgorithm); |
| 480 | + flushAlgorithm, |
| 481 | + cancelAlgorithm); |
473 | 482 | } |
474 | 483 |
|
475 | 484 | function transformStreamDefaultControllerClearAlgorithms(controller) { |
476 | 485 | controller[kState].transformAlgorithm = undefined; |
477 | 486 | controller[kState].flushAlgorithm = undefined; |
| 487 | + controller[kState].cancelAlgorithm = undefined; |
478 | 488 | } |
479 | 489 |
|
480 | 490 | function transformStreamDefaultControllerEnqueue(controller, chunk) { |
@@ -563,7 +573,40 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { |
563 | 573 | } |
564 | 574 |
|
565 | 575 | async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { |
566 | | - transformStreamError(stream, reason); |
| 576 | + const { |
| 577 | + controller, |
| 578 | + readable, |
| 579 | + } = stream[kState]; |
| 580 | + |
| 581 | + if (controller[kState].finishPromise !== undefined) { |
| 582 | + return controller[kState].finishPromise; |
| 583 | + } |
| 584 | + |
| 585 | + const { promise, resolve, reject } = createDeferredPromise(); |
| 586 | + controller[kState].finishPromise = promise; |
| 587 | + const cancelPromise = ensureIsPromise( |
| 588 | + controller[kState].cancelAlgorithm, |
| 589 | + controller, |
| 590 | + reason); |
| 591 | + transformStreamDefaultControllerClearAlgorithms(controller); |
| 592 | + |
| 593 | + PromisePrototypeThen( |
| 594 | + cancelPromise, |
| 595 | + () => { |
| 596 | + if (readable[kState].state === 'errored') |
| 597 | + reject(readable[kState].storedError); |
| 598 | + else { |
| 599 | + readableStreamDefaultControllerError(readable[kState].controller, reason); |
| 600 | + resolve(); |
| 601 | + } |
| 602 | + }, |
| 603 | + (error) => { |
| 604 | + readableStreamDefaultControllerError(readable[kState].controller, error); |
| 605 | + reject(error); |
| 606 | + }, |
| 607 | + ); |
| 608 | + |
| 609 | + return controller[kState].finishPromise; |
567 | 610 | } |
568 | 611 |
|
569 | 612 | function transformStreamDefaultSinkCloseAlgorithm(stream) { |
@@ -572,23 +615,32 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { |
572 | 615 | controller, |
573 | 616 | } = stream[kState]; |
574 | 617 |
|
| 618 | + if (controller[kState].finishPromise !== undefined) { |
| 619 | + return controller[kState].finishPromise; |
| 620 | + } |
| 621 | + const { promise, resolve, reject } = createDeferredPromise(); |
| 622 | + controller[kState].finishPromise = promise; |
575 | 623 | const flushPromise = |
576 | 624 | ensureIsPromise( |
577 | 625 | controller[kState].flushAlgorithm, |
578 | 626 | controller, |
579 | 627 | controller); |
580 | 628 | transformStreamDefaultControllerClearAlgorithms(controller); |
581 | | - return PromisePrototypeThen( |
| 629 | + PromisePrototypeThen( |
582 | 630 | flushPromise, |
583 | 631 | () => { |
584 | 632 | if (readable[kState].state === 'errored') |
585 | | - throw readable[kState].storedError; |
586 | | - readableStreamDefaultControllerClose(readable[kState].controller); |
| 633 | + reject(readable[kState].storedError); |
| 634 | + else { |
| 635 | + readableStreamDefaultControllerClose(readable[kState].controller); |
| 636 | + resolve(); |
| 637 | + } |
587 | 638 | }, |
588 | 639 | (error) => { |
589 | | - transformStreamError(stream, error); |
590 | | - throw readable[kState].storedError; |
| 640 | + readableStreamDefaultControllerError(readable[kState].controller, error); |
| 641 | + reject(error); |
591 | 642 | }); |
| 643 | + return controller[kState].finishPromise; |
592 | 644 | } |
593 | 645 |
|
594 | 646 | function transformStreamDefaultSourcePullAlgorithm(stream) { |
@@ -598,6 +650,48 @@ function transformStreamDefaultSourcePullAlgorithm(stream) { |
598 | 650 | return stream[kState].backpressureChange.promise; |
599 | 651 | } |
600 | 652 |
|
| 653 | +function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { |
| 654 | + const { |
| 655 | + controller, |
| 656 | + writable, |
| 657 | + } = stream[kState]; |
| 658 | + |
| 659 | + if (controller[kState].finishPromise !== undefined) { |
| 660 | + return controller[kState].finishPromise; |
| 661 | + } |
| 662 | + |
| 663 | + const { promise, resolve, reject } = createDeferredPromise(); |
| 664 | + controller[kState].finishPromise = promise; |
| 665 | + const cancelPromise = ensureIsPromise( |
| 666 | + controller[kState].cancelAlgorithm, |
| 667 | + controller, |
| 668 | + reason); |
| 669 | + transformStreamDefaultControllerClearAlgorithms(controller); |
| 670 | + |
| 671 | + PromisePrototypeThen(cancelPromise, |
| 672 | + () => { |
| 673 | + if (writable[kState].state === 'errored') |
| 674 | + reject(writable[kState].storedError); |
| 675 | + else { |
| 676 | + writableStreamDefaultControllerErrorIfNeeded( |
| 677 | + writable[kState].controller, |
| 678 | + reason); |
| 679 | + transformStreamUnblockWrite(stream); |
| 680 | + resolve(); |
| 681 | + } |
| 682 | + }, |
| 683 | + (error) => { |
| 684 | + writableStreamDefaultControllerErrorIfNeeded( |
| 685 | + writable[kState].controller, |
| 686 | + error); |
| 687 | + transformStreamUnblockWrite(stream); |
| 688 | + reject(error); |
| 689 | + }, |
| 690 | + ); |
| 691 | + |
| 692 | + return controller[kState].finishPromise; |
| 693 | +} |
| 694 | + |
601 | 695 | module.exports = { |
602 | 696 | TransformStream, |
603 | 697 | TransformStreamDefaultController, |
|