From 841d2fdf0caf092899fe9d2ec476da4b7ef882a1 Mon Sep 17 00:00:00 2001 From: Youenn Fablet Date: Sun, 9 Apr 2023 13:58:50 +0200 Subject: [PATCH] Add support for a new ReadableStream "owning" type. --- .github/workflows/test.yml | 2 +- index.bs | 101 ++++++++++++------ .../lib/ReadableStream-impl.js | 2 +- .../ReadableStreamDefaultController-impl.js | 5 +- .../ReadableStreamDefaultController.webidl | 6 +- .../lib/UnderlyingSource.webidl | 2 +- .../lib/abstract-ops/miscellaneous.js | 4 + .../lib/abstract-ops/queue-with-sizes.js | 20 +++- .../lib/abstract-ops/readable-streams.js | 32 ++++-- .../lib/abstract-ops/transform-streams.js | 2 +- .../lib/abstract-ops/writable-streams.js | 4 +- .../run-web-platform-tests.js | 5 +- reference-implementation/web-platform-tests | 2 +- 13 files changed, 132 insertions(+), 55 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 25ede8c9d..f22650452 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -19,6 +19,6 @@ jobs: submodules: true - uses: actions/setup-node@v1 with: - node-version: 14 + node-version: 19 - run: npm install - run: npm test diff --git a/index.bs b/index.bs index da6a50aed..286943bca 100644 --- a/index.bs +++ b/index.bs @@ -19,6 +19,7 @@ spec:infra; type:dfn; text:list spec:html; type:dfn; text:entangle spec:html; type:dfn; text:message port post message steps spec:html; type:dfn; text:port message queue +spec:html; type:dfn; text:transferable objects
@@ -571,7 +572,7 @@ callback UnderlyingSourceStartCallback = any (ReadableStreamController controlle
 callback UnderlyingSourcePullCallback = Promise (ReadableStreamController controller);
 callback UnderlyingSourceCancelCallback = Promise (optional any reason);
 
-enum ReadableStreamType { "bytes" };
+enum ReadableStreamType { "bytes", "owning" };
 
 
 
@@ -639,8 +640,7 @@ enum ReadableStreamType { "bytes" }; something more persistently wrong. -
type (byte streams - only)
+
type

Can be set to "bytes" to signal that the constructed {{ReadableStream}} is a readable byte stream. This ensures that the resulting @@ -651,8 +651,16 @@ enum ReadableStreamType { "bytes" };

For an example of how to set up a readable byte stream, including using the different controller interface, see [[#example-rbs-push]]. -

Setting any value other than "{{ReadableStreamType/bytes}}" or undefined will cause the - {{ReadableStream()}} constructor to throw an exception. +

Can be set to "owning" to signal that the + constructed {{ReadableStream}} will own chunks (via transfer or serialization) before enqueuing them. + This ensures that enqueued chunks are not mutable by the source. + Transferred or serialized chunks may have closing steps which are executed if + enqueued chunks are dequeued without being provided to the application, for instance when + a {{ReadableStream}} is errored. +

+ +

Setting any value other than "{{ReadableStreamType/bytes}}", "{{ReadableStreamType/owning}}" + or undefined will cause the {{ReadableStream()}} constructor to throw an exception.

autoAllocateChunkSize (byte streams only)
@@ -801,7 +809,8 @@ option. If {{UnderlyingSource/type}} is set to undefined (including via omission 1. Perform ? [$SetUpReadableByteStreamControllerFromUnderlyingSource$]([=this=], |underlyingSource|, |underlyingSourceDict|, |highWaterMark|). 1. Otherwise, - 1. Assert: |underlyingSourceDict|["{{UnderlyingSource/type}}"] does not [=map/exist=]. + 1. Assert: |underlyingSourceDict|["{{UnderlyingSource/type}}"] does not [=map/exist=] or + is "{{ReadableStreamType/owning}}". 1. Let |sizeAlgorithm| be ! [$ExtractSizeAlgorithm$](|strategy|). 1. Let |highWaterMark| be ? [$ExtractHighWaterMark$](|strategy|, 1). 1. Perform ? [$SetUpReadableStreamDefaultControllerFromUnderlyingSource$]([=this=], @@ -1461,7 +1470,7 @@ interface ReadableStreamDefaultController { readonly attribute unrestricted double? desiredSize; undefined close(); - undefined enqueue(optional any chunk); + undefined enqueue(optional any chunk, optional StructuredSerializeOptions options = { }); undefined error(optional any e); }; @@ -1523,6 +1532,10 @@ the following table: \[[stream]] The {{ReadableStream}} instance controlled + + \[[isOwning]] + A boolean flag indicating whether to take ownership of enqueued chunks + via transfer or serialization.

Methods and properties

@@ -1569,11 +1582,12 @@ the following table:
The enqueue(|chunk|) method steps are: + for="ReadableStreamDefaultController">enqueue(|chunk|, |options|) method steps are: + 1. Let |transferList| be |options|["transfer"]. 1. If ! [$ReadableStreamDefaultControllerCanCloseOrEnqueue$]([=this=]) is false, throw a {{TypeError}} exception. - 1. Perform ? [$ReadableStreamDefaultControllerEnqueue$]([=this=], |chunk|). + 1. Perform ? [$ReadableStreamDefaultControllerEnqueue$]([=this=], |chunk|, |transferList|).
@@ -2235,8 +2249,8 @@ create them does not matter. objects|transferring=] their [=chunks=]. However, it does introduce a noticeable asymmetry between the two branches, and limits the possible [=chunks=] to serializable ones. [[!HTML]] - If |stream| is a [=readable byte stream=], then |cloneForBranch2| is ignored and chunks are cloned - unconditionally. + If |stream| is a [=readable byte stream=], or if |stream| type is "{{ReadableStreamType/owning}}", + then |cloneForBranch2| is ignored and chunks are cloned unconditionally.

In this standard ReadableStreamTee is always called with |cloneForBranch2| set to false; other specifications pass true via the [=ReadableStream/tee=] wrapper algorithm. @@ -2247,6 +2261,8 @@ create them does not matter. 1. Assert: |cloneForBranch2| is a boolean. 1. If |stream|.[=ReadableStream/[[controller]]=] [=implements=] {{ReadableByteStreamController}}, return ? [$ReadableByteStreamTee$](|stream|). + 1. If |stream|.[=ReadableStream/[[controller]]=].[=ReadableStreamDefaultController/[[isOwning]]=] + is true, return ? [$ReadableStreamDefaultTee$](|stream|, true). 1. Return ? [$ReadableStreamDefaultTee$](|stream|, |cloneForBranch2|).

@@ -2287,10 +2303,10 @@ create them does not matter. 1. Otherwise, set |chunk2| to |cloneResult|.\[[Value]]. 1. If |canceled1| is false, perform ! [$ReadableStreamDefaultControllerEnqueue$](|branch1|.[=ReadableStream/[[controller]]=], - |chunk1|). + |chunk1|, undefined). 1. If |canceled2| is false, perform ! [$ReadableStreamDefaultControllerEnqueue$](|branch2|.[=ReadableStream/[[controller]]=], - |chunk2|). + |chunk2|, undefined). 1. Set |reading| to false. 1. If |readAgain| is true, perform |pullAlgorithm|. @@ -2950,13 +2966,19 @@ The following abstract operations support the implementation of the
ReadableStreamDefaultControllerEnqueue(|controller|, - |chunk|) performs the following steps: + |chunk|, |transferList|) performs the following steps: 1. If ! [$ReadableStreamDefaultControllerCanCloseOrEnqueue$](|controller|) is false, return. 1. Let |stream| be |controller|.[=ReadableStreamDefaultController/[[stream]]=]. 1. If ! [$IsReadableStreamLocked$](|stream|) is true and ! - [$ReadableStreamGetNumReadRequests$](|stream|) > 0, perform ! - [$ReadableStreamFulfillReadRequest$](|stream|, |chunk|, false). + [$ReadableStreamGetNumReadRequests$](|stream|) > 0, perform the following steps: + 1. Let |internalChunk| be |chunk|. + 1. If |controller|.[=ReadableStreamDefaultController/[[isOwning]]=] is true, perform the following steps: + 1. Set |internalChunk| to [$StructuredTransferOrClone$](|chunk|, |transferList|). + 1. If |internalChunk| is an abrupt completion, + 1. Perform ! [$ReadableStreamDefaultControllerError$](|controller|, |internalChunk|.\[[Value]]). + 1. Return |internalChunk|. + 1. Perform ! [$ReadableStreamFulfillReadRequest$](|stream|, |internalChunk|, false). 1. Otherwise, 1. Let |result| be the result of performing |controller|.[=ReadableStreamDefaultController/[[strategySizeAlgorithm]]=], passing in |chunk|, @@ -2965,7 +2987,7 @@ The following abstract operations support the implementation of the 1. Perform ! [$ReadableStreamDefaultControllerError$](|controller|, |result|.\[[Value]]). 1. Return |result|. 1. Let |chunkSize| be |result|.\[[Value]]. - 1. Let |enqueueResult| be [$EnqueueValueWithSize$](|controller|, |chunk|, |chunkSize|). + 1. Let |enqueueResult| be [$EnqueueValueWithSize$](|controller|, |chunk|, |chunkSize|, |transferList|). 1. If |enqueueResult| is an abrupt completion, 1. Perform ! [$ReadableStreamDefaultControllerError$](|controller|, |enqueueResult|.\[[Value]]). 1. Return |enqueueResult|. @@ -3029,7 +3051,7 @@ The following abstract operations support the implementation of the SetUpReadableStreamDefaultController(|stream|, |controller|, |startAlgorithm|, |pullAlgorithm|, |cancelAlgorithm|, |highWaterMark|, - |sizeAlgorithm|) performs the following steps: + |sizeAlgorithm|, |isOwning|) performs the following steps: 1. Assert: |stream|.[=ReadableStream/[[controller]]=] is undefined. 1. Set |controller|.[=ReadableStreamDefaultController/[[stream]]=] to |stream|. @@ -3039,8 +3061,9 @@ The following abstract operations support the implementation of the |controller|.[=ReadableStreamDefaultController/[[pullAgain]]=], and |controller|.[=ReadableStreamDefaultController/[[pulling]]=] to false. 1. Set |controller|.[=ReadableStreamDefaultController/[[strategySizeAlgorithm]]=] to - |sizeAlgorithm| and |controller|.[=ReadableStreamDefaultController/[[strategyHWM]]=] to - |highWaterMark|. + |sizeAlgorithm|, |controller|.[=ReadableStreamDefaultController/[[strategyHWM]]=] to + |highWaterMark| and |controller|.[=ReadableStreamDefaultController/[[isOwning]]=] to + |isOwning|. 1. Set |controller|.[=ReadableStreamDefaultController/[[pullAlgorithm]]=] to |pullAlgorithm|. 1. Set |controller|.[=ReadableStreamDefaultController/[[cancelAlgorithm]]=] to |cancelAlgorithm|. 1. Set |stream|.[=ReadableStream/[[controller]]=] to |controller|. @@ -3065,6 +3088,8 @@ The following abstract operations support the implementation of the 1. Let |startAlgorithm| be an algorithm that returns undefined. 1. Let |pullAlgorithm| be an algorithm that returns [=a promise resolved with=] undefined. 1. Let |cancelAlgorithm| be an algorithm that returns [=a promise resolved with=] undefined. + 1. Let |isOwning| be true if |underlyingSourceDict|["{{UnderlyingSource/type}}"] is + "{{ReadableStreamType/owning}}" and false otherwise. 1. If |underlyingSourceDict|["{{UnderlyingSource/start}}"] [=map/exists=], then set |startAlgorithm| to an algorithm which returns the result of [=invoking=] |underlyingSourceDict|["{{UnderlyingSource/start}}"] with argument list @@ -3078,7 +3103,7 @@ The following abstract operations support the implementation of the [=invoking=] |underlyingSourceDict|["{{UnderlyingSource/cancel}}"] with argument list « |reason| » and [=callback this value=] |underlyingSource|. 1. Perform ? [$SetUpReadableStreamDefaultController$](|stream|, |controller|, |startAlgorithm|, - |pullAlgorithm|, |cancelAlgorithm|, |highWaterMark|, |sizeAlgorithm|). + |pullAlgorithm|, |cancelAlgorithm|, |highWaterMark|, |sizeAlgorithm|, |isOwning|).

Byte stream controllers

@@ -5186,7 +5211,7 @@ The following abstract operations support the implementation of the id="writable-stream-default-controller-close">WritableStreamDefaultControllerClose(|controller|) performs the following steps: - 1. Perform ! [$EnqueueValueWithSize$](|controller|, [=close sentinel=], 0). + 1. Perform ! [$EnqueueValueWithSize$](|controller|, [=close sentinel=], 0, undefined). 1. Perform ! [$WritableStreamDefaultControllerAdvanceQueueIfNeeded$](|controller|). @@ -5290,7 +5315,7 @@ The following abstract operations support the implementation of the id="writable-stream-default-controller-write">WritableStreamDefaultControllerWrite(|controller|, |chunk|, |chunkSize|) performs the following steps: - 1. Let |enqueueResult| be [$EnqueueValueWithSize$](|controller|, |chunk|, |chunkSize|). + 1. Let |enqueueResult| be [$EnqueueValueWithSize$](|controller|, |chunk|, |chunkSize|, undefined). 1. If |enqueueResult| is an abrupt completion, 1. Perform ! [$WritableStreamDefaultControllerErrorIfNeeded$](|controller|, |enqueueResult|.\[[Value]]). @@ -5884,7 +5909,7 @@ The following abstract operations support the implementaiton of the 1. If ! [$ReadableStreamDefaultControllerCanCloseOrEnqueue$](|readableController|) is false, throw a {{TypeError}} exception. 1. Let |enqueueResult| be [$ReadableStreamDefaultControllerEnqueue$](|readableController|, - |chunk|). + |chunk|, undefined). 1. If |enqueueResult| is an abrupt completion, 1. Perform ! [$TransformStreamErrorWritableAndUnblockWrite$](|stream|, |enqueueResult|.\[[Value]]). @@ -6362,13 +6387,17 @@ for="value-with-size">value and size.
EnqueueValueWithSize(|container|, |value|, |size|) performs the + id="enqueue-value-with-size">EnqueueValueWithSize(|container|, |value|, |size|, |transferList|) performs the following steps: 1. Assert: |container| has \[[queue]] and \[[queueTotalSize]] internal slots. 1. If ! [$IsNonNegativeNumber$](|size|) is false, throw a {{RangeError}} exception. 1. If |size| is +∞, throw a {{RangeError}} exception. - 1. [=list/Append=] a new [=value-with-size=] with [=value-with-size/value=] |value| and + 1. Let |enqueuedValue| be |value|. + 1. If |container| has a \[[isOwning]] internal slot whose value is true, perform the following steps: + 1. Set |enqueuedValue| to [$StructuredTransferOrClone$](|value|, |transferList|). + 1. If |enqueuedValue| is an abrupt completion, return |enqueuedValue|. + 1. [=list/Append=] a new [=value-with-size=] with [=value-with-size/value=] |enqueuedValue| and [=value-with-size/size=] |size| to |container|.\[[queue]]. 1. Set |container|.\[[queueTotalSize]] to |container|.\[[queueTotalSize]] + |size|.
@@ -6388,6 +6417,10 @@ for="value-with-size">value and size. performs the following steps: 1. Assert: |container| has \[[queue]] and \[[queueTotalSize]] internal slots. + 1. If |container| has a \[[isOwning]] internal slot whose value is true, perform the following steps until |container|.\[[queue]] + is [=list/is empty|empty=]: + 1. Let |chunk| be ! [$DequeueValue$]([=this=]). + 1. If |chunk| has [=closing steps=], perform the [=closing steps=] given |chunk|. 1. Set |container|.\[[queue]] to a new empty [=list=]. 1. Set |container|.\[[queueTotalSize]] to 0. @@ -6447,7 +6480,7 @@ abstract operations are used to implement these "cross-realm transforms". 1. Let |value| be ! [$Get$](|data|, "`value`"). 1. Assert: [$Type$](|type|) is String. 1. If |type| is "`chunk`", - 1. Perform ! [$ReadableStreamDefaultControllerEnqueue$](|controller|, |value|). + 1. Perform ! [$ReadableStreamDefaultControllerEnqueue$](|controller|, |value|, undefined). 1. Otherwise, if |type| is "`close`", 1. Perform ! [$ReadableStreamDefaultControllerClose$](|controller|). 1. Disentangle |port|. @@ -6596,6 +6629,14 @@ The following abstract operations are a grab-bag of utilities. 1. Return ? [$StructuredDeserialize$](|serialized|, [=the current Realm=]). +
+ StructuredTransferOrClone(|value|, |transferList|) + performs the following steps: + 1. Let |serialized| be ! [$StructuredSerializeWithTransfer$](|value|, |transferList|). + 1. Let |deserialized| be ! [$StructuredDeserializeWithTransfer$](|serialized|, [=the current Realm=]). + 1. return |deserialized|.\[[Deserialized]]. +
+

Using streams in other specifications

Much of this standard concerns itself with the internal machinery of streams. Other specifications @@ -6727,13 +6768,13 @@ mark=] is greater than zero.
- To enqueue the JavaScript value |chunk| into a - {{ReadableStream}} |stream|: + To enqueue the JavaScript value |chunk| + with an optional |transferList| into a {{ReadableStream}} |stream|: 1. If |stream|.[=ReadableStream/[[controller]]=] [=implements=] {{ReadableStreamDefaultController}}, 1. Perform ! [$ReadableStreamDefaultControllerEnqueue$](|stream|.[=ReadableStream/[[controller]]=], - |chunk|). + |chunk|, |transferList|). 1. Otherwise, 1. Assert: |stream|.[=ReadableStream/[[controller]]=] [=implements=] {{ReadableByteStreamController}}. diff --git a/reference-implementation/lib/ReadableStream-impl.js b/reference-implementation/lib/ReadableStream-impl.js index 1eda283e1..47bde08b7 100644 --- a/reference-implementation/lib/ReadableStream-impl.js +++ b/reference-implementation/lib/ReadableStream-impl.js @@ -29,7 +29,7 @@ exports.implementation = class ReadableStreamImpl { this, underlyingSource, underlyingSourceDict, highWaterMark ); } else { - assert(!('type' in underlyingSourceDict)); + assert(!('type' in underlyingSourceDict) || underlyingSourceDict.type === 'owning'); const sizeAlgorithm = ExtractSizeAlgorithm(strategy); const highWaterMark = ExtractHighWaterMark(strategy, 1); aos.SetUpReadableStreamDefaultControllerFromUnderlyingSource( diff --git a/reference-implementation/lib/ReadableStreamDefaultController-impl.js b/reference-implementation/lib/ReadableStreamDefaultController-impl.js index 5c7ec7033..d29faf89b 100644 --- a/reference-implementation/lib/ReadableStreamDefaultController-impl.js +++ b/reference-implementation/lib/ReadableStreamDefaultController-impl.js @@ -17,12 +17,13 @@ exports.implementation = class ReadableStreamDefaultControllerImpl { aos.ReadableStreamDefaultControllerClose(this); } - enqueue(chunk) { + enqueue(chunk, options) { + const transferList = options ? options.transfer : undefined; if (aos.ReadableStreamDefaultControllerCanCloseOrEnqueue(this) === false) { throw new TypeError('The stream is not in a state that permits enqueue'); } - return aos.ReadableStreamDefaultControllerEnqueue(this, chunk); + return aos.ReadableStreamDefaultControllerEnqueue(this, chunk, transferList); } error(e) { diff --git a/reference-implementation/lib/ReadableStreamDefaultController.webidl b/reference-implementation/lib/ReadableStreamDefaultController.webidl index aeea7249f..dc02cfdcb 100644 --- a/reference-implementation/lib/ReadableStreamDefaultController.webidl +++ b/reference-implementation/lib/ReadableStreamDefaultController.webidl @@ -1,8 +1,12 @@ +dictionary StructuredSerializeOptions { + sequence transfer = []; +}; + [Exposed=(Window,Worker,Worklet)] interface ReadableStreamDefaultController { readonly attribute unrestricted double? desiredSize; undefined close(); - undefined enqueue(optional any chunk); + undefined enqueue(optional any chunk, optional StructuredSerializeOptions options = { }); undefined error(optional any e); }; diff --git a/reference-implementation/lib/UnderlyingSource.webidl b/reference-implementation/lib/UnderlyingSource.webidl index 7a0047638..b8aa75b70 100644 --- a/reference-implementation/lib/UnderlyingSource.webidl +++ b/reference-implementation/lib/UnderlyingSource.webidl @@ -12,4 +12,4 @@ callback UnderlyingSourceStartCallback = any (ReadableStreamController controlle callback UnderlyingSourcePullCallback = Promise (ReadableStreamController controller); callback UnderlyingSourceCancelCallback = Promise (optional any reason); -enum ReadableStreamType { "bytes" }; +enum ReadableStreamType { "bytes", "owning" }; diff --git a/reference-implementation/lib/abstract-ops/miscellaneous.js b/reference-implementation/lib/abstract-ops/miscellaneous.js index 08589a740..ed55a5268 100644 --- a/reference-implementation/lib/abstract-ops/miscellaneous.js +++ b/reference-implementation/lib/abstract-ops/miscellaneous.js @@ -20,3 +20,7 @@ exports.CloneAsUint8Array = O => { const buffer = O.buffer.slice(O.byteOffset, O.byteOffset + O.byteLength); return new Uint8Array(buffer); }; + +exports.StructuredTransferOrClone = (value, transferList) => { + return globalThis.structuredClone(value, { transfer: transferList }); +}; diff --git a/reference-implementation/lib/abstract-ops/queue-with-sizes.js b/reference-implementation/lib/abstract-ops/queue-with-sizes.js index 22086caa5..8580f8021 100644 --- a/reference-implementation/lib/abstract-ops/queue-with-sizes.js +++ b/reference-implementation/lib/abstract-ops/queue-with-sizes.js @@ -1,6 +1,6 @@ 'use strict'; const assert = require('assert'); -const { IsNonNegativeNumber } = require('./miscellaneous.js'); +const { IsNonNegativeNumber, StructuredTransferOrClone } = require('./miscellaneous.js'); exports.DequeueValue = container => { assert('_queue' in container && '_queueTotalSize' in container); @@ -15,7 +15,7 @@ exports.DequeueValue = container => { return pair.value; }; -exports.EnqueueValueWithSize = (container, value, size) => { +exports.EnqueueValueWithSize = (container, value, size, transferList) => { assert('_queue' in container && '_queueTotalSize' in container); if (!IsNonNegativeNumber(size)) { @@ -24,7 +24,9 @@ exports.EnqueueValueWithSize = (container, value, size) => { if (size === Infinity) { throw new RangeError('Size must be a finite, non-NaN, non-negative number.'); } - + if (container.isOwning) { + value = StructuredTransferOrClone(value, transferList); + } container._queue.push({ value, size }); container._queueTotalSize += size; }; @@ -40,6 +42,18 @@ exports.PeekQueueValue = container => { exports.ResetQueue = container => { assert('_queue' in container && '_queueTotalSize' in container); + if (container.isOwning) { + while (container._queue.length > 0) { + const value = exports.DequeueValue(container); + if (typeof value.close === 'function') { + try { + value.close(); + } catch (closeException) { + // Nothing to do. + } + } + } + } container._queue = []; container._queueTotalSize = 0; }; diff --git a/reference-implementation/lib/abstract-ops/readable-streams.js b/reference-implementation/lib/abstract-ops/readable-streams.js index db1da4c73..9080724ae 100644 --- a/reference-implementation/lib/abstract-ops/readable-streams.js +++ b/reference-implementation/lib/abstract-ops/readable-streams.js @@ -6,7 +6,7 @@ const { promiseResolvedWith, promiseRejectedWith, newPromise, resolvePromise, re require('../helpers/webidl.js'); const { CanTransferArrayBuffer, CopyDataBlockBytes, CreateArrayFromList, IsDetachedBuffer, TransferArrayBuffer } = require('./ecmascript.js'); -const { CloneAsUint8Array, IsNonNegativeNumber } = require('./miscellaneous.js'); +const { CloneAsUint8Array, IsNonNegativeNumber, StructuredTransferOrClone } = require('./miscellaneous.js'); const { EnqueueValueWithSize, ResetQueue } = require('./queue-with-sizes.js'); const { AcquireWritableStreamDefaultWriter, IsWritableStreamLocked, WritableStreamAbort, WritableStreamDefaultWriterCloseWithErrorPropagation, WritableStreamDefaultWriterRelease, @@ -89,7 +89,7 @@ function CreateReadableStream(startAlgorithm, pullAlgorithm, cancelAlgorithm, hi const controller = ReadableStreamDefaultController.new(globalThis); SetUpReadableStreamDefaultController( - stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm + stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm, false ); return stream; @@ -340,7 +340,7 @@ function ReadableStreamTee(stream, cloneForBranch2) { if (ReadableByteStreamController.isImpl(stream._controller)) { return ReadableByteStreamTee(stream); } - return ReadableStreamDefaultTee(stream, cloneForBranch2); + return ReadableStreamDefaultTee(stream, stream._controller.isOwning ? true : cloneForBranch2); } function ReadableStreamDefaultTee(stream, cloneForBranch2) { @@ -392,10 +392,10 @@ function ReadableStreamDefaultTee(stream, cloneForBranch2) { // } if (canceled1 === false) { - ReadableStreamDefaultControllerEnqueue(branch1._controller, chunk1); + ReadableStreamDefaultControllerEnqueue(branch1._controller, chunk1, undefined); } if (canceled2 === false) { - ReadableStreamDefaultControllerEnqueue(branch2._controller, chunk2); + ReadableStreamDefaultControllerEnqueue(branch2._controller, chunk2, undefined); } reading = false; @@ -1074,7 +1074,7 @@ function ReadableStreamDefaultControllerClose(controller) { } } -function ReadableStreamDefaultControllerEnqueue(controller, chunk) { +function ReadableStreamDefaultControllerEnqueue(controller, chunk, transferList) { if (ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) === false) { return; } @@ -1082,6 +1082,14 @@ function ReadableStreamDefaultControllerEnqueue(controller, chunk) { const stream = controller._stream; if (IsReadableStreamLocked(stream) === true && ReadableStreamGetNumReadRequests(stream) > 0) { + if (controller.isOwning) { + try { + chunk = StructuredTransferOrClone(chunk, transferList); + } catch (chunkCloneError) { + ReadableStreamDefaultControllerError(controller, chunkCloneError); + throw chunkCloneError; + } + } ReadableStreamFulfillReadRequest(stream, chunk, false); } else { let chunkSize; @@ -1093,7 +1101,7 @@ function ReadableStreamDefaultControllerEnqueue(controller, chunk) { } try { - EnqueueValueWithSize(controller, chunk, chunkSize); + EnqueueValueWithSize(controller, chunk, chunkSize, transferList); } catch (enqueueE) { ReadableStreamDefaultControllerError(controller, enqueueE); throw enqueueE; @@ -1148,7 +1156,7 @@ function ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) { } function SetUpReadableStreamDefaultController( - stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm) { + stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm, isOwning) { assert(stream._controller === undefined); controller._stream = stream; @@ -1169,6 +1177,8 @@ function SetUpReadableStreamDefaultController( controller._pullAlgorithm = pullAlgorithm; controller._cancelAlgorithm = cancelAlgorithm; + controller.isOwning = isOwning; + stream._controller = controller; const startResult = startAlgorithm(); @@ -1195,7 +1205,7 @@ function SetUpReadableStreamDefaultControllerFromUnderlyingSource( let startAlgorithm = () => undefined; let pullAlgorithm = () => promiseResolvedWith(undefined); let cancelAlgorithm = () => promiseResolvedWith(undefined); - + const isOwning = underlyingSourceDict.type === 'owning'; if ('start' in underlyingSourceDict) { startAlgorithm = () => underlyingSourceDict.start.call(underlyingSource, controller); } @@ -1207,8 +1217,8 @@ function SetUpReadableStreamDefaultControllerFromUnderlyingSource( } SetUpReadableStreamDefaultController( - stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm - ); + stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm, + isOwning); } // Byte stream controllers diff --git a/reference-implementation/lib/abstract-ops/transform-streams.js b/reference-implementation/lib/abstract-ops/transform-streams.js index 8e3f5fcc3..1152d19de 100644 --- a/reference-implementation/lib/abstract-ops/transform-streams.js +++ b/reference-implementation/lib/abstract-ops/transform-streams.js @@ -155,7 +155,7 @@ function TransformStreamDefaultControllerEnqueue(controller, chunk) { // accept TransformStreamDefaultControllerEnqueue() calls. try { - ReadableStreamDefaultControllerEnqueue(readableController, chunk); + ReadableStreamDefaultControllerEnqueue(readableController, chunk, undefined); } catch (e) { // This happens when readableStrategy.size() throws. TransformStreamErrorWritableAndUnblockWrite(stream, e); diff --git a/reference-implementation/lib/abstract-ops/writable-streams.js b/reference-implementation/lib/abstract-ops/writable-streams.js index cf303bfe7..88eef0888 100644 --- a/reference-implementation/lib/abstract-ops/writable-streams.js +++ b/reference-implementation/lib/abstract-ops/writable-streams.js @@ -637,7 +637,7 @@ function WritableStreamDefaultControllerClearAlgorithms(controller) { } function WritableStreamDefaultControllerClose(controller) { - EnqueueValueWithSize(controller, closeSentinel, 0); + EnqueueValueWithSize(controller, closeSentinel, 0, undefined); WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); } @@ -729,7 +729,7 @@ function WritableStreamDefaultControllerProcessWrite(controller, chunk) { function WritableStreamDefaultControllerWrite(controller, chunk, chunkSize) { try { - EnqueueValueWithSize(controller, chunk, chunkSize); + EnqueueValueWithSize(controller, chunk, chunkSize, undefined); } catch (enqueueE) { WritableStreamDefaultControllerErrorIfNeeded(controller, enqueueE); return; diff --git a/reference-implementation/run-web-platform-tests.js b/reference-implementation/run-web-platform-tests.js index 836911eeb..8d857c9f7 100644 --- a/reference-implementation/run-web-platform-tests.js +++ b/reference-implementation/run-web-platform-tests.js @@ -35,7 +35,9 @@ async function main() { const excludeGlobs = [ // These tests use ArrayBuffers backed by WebAssembly.Memory objects, which *should* be non-transferable. // However, our TransferArrayBuffer implementation cannot detect these, and will incorrectly "transfer" them anyway. - 'readable-byte-streams/non-transferable-buffers.any.html' + 'readable-byte-streams/non-transferable-buffers.any.html', + 'readable-streams/owning-type-message-port.any.html', // MessagePort is not defined. + 'readable-streams/owning-type-video-frame.any.html' // VideoFrame is not defined. ]; const anyTestPattern = /\.any\.html$/; @@ -58,6 +60,7 @@ async function main() { } }; }; + window.structuredClone = globalThis.structuredClone; window.eval(bundledJS); }, filter(testPath) { diff --git a/reference-implementation/web-platform-tests b/reference-implementation/web-platform-tests index 6a46d9cb8..35ad0cf07 160000 --- a/reference-implementation/web-platform-tests +++ b/reference-implementation/web-platform-tests @@ -1 +1 @@ -Subproject commit 6a46d9cb8d20c510a620141c721b81b460a4ee55 +Subproject commit 35ad0cf07278adce002370d29d94aeb1da83e537