diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7102b5b04..58c4b8177 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -17,7 +17,7 @@ jobs: - uses: actions/checkout@v2 with: submodules: true - - uses: actions/setup-node@v1 + - uses: actions/setup-node@v3 with: node-version: 18 - run: npm install diff --git a/index.bs b/index.bs index 7202f89a8..a41a6431a 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
@@ -593,7 +594,7 @@ callback UnderlyingSourceStartCallback = any (ReadableStreamController controlle
 callback UnderlyingSourcePullCallback = Promise (ReadableStreamController controller);
 callback UnderlyingSourceCancelCallback = Promise (optional any reason);
 
-enum ReadableStreamType { "bytes" };
+enum ReadableStreamType { "bytes", "owning" };
 
 
 
@@ -661,8 +662,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 @@ -673,8 +673,15 @@ 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 cannot be mutated by the source. + Chunks may have dispose 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)
@@ -831,7 +838,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=], @@ -1518,7 +1526,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); }; @@ -1543,6 +1551,10 @@ the following table: A boolean flag indicating whether the stream has been closed by its [=underlying source=], but still has [=chunks=] in its internal queue that have not yet been read + + \[[isOwning]] + A boolean flag indicating whether to take ownership of enqueued chunks + via transfer or serialization. \[[pullAgain]] A boolean flag set to true if the stream's mechanisms requested a call @@ -1598,9 +1610,10 @@ the following table: previously-enqueued [=chunks=] from the stream, but once those are read, the stream will become closed. -
controller.{{ReadableStreamDefaultController/enqueue()|enqueue}}(chunk) +
controller.{{ReadableStreamDefaultController/enqueue()|enqueue}}(|chunk|, |options|)
-

Enqueues the given [=chunk=] chunk in the controlled readable stream. +

Enqueues the given [=chunk=] |chunk| in the controlled readable stream, + with |options|.

controller.{{ReadableStreamDefaultController/error()|error}}(e)
@@ -1626,11 +1639,14 @@ 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 |transferList| is not [=list/is empty|empty=] and [=this=].[=ReadableStreamDefaultController/[[isOwning]]=] is false, + throw a {{TypeError}} exception. 1. If ! [$ReadableStreamDefaultControllerCanCloseOrEnqueue$]([=this=]) is false, throw a {{TypeError}} exception. - 1. Perform ? [$ReadableStreamDefaultControllerEnqueue$]([=this=], |chunk|). + 1. Perform ? [$ReadableStreamDefaultControllerEnqueue$]([=this=], |chunk|, |transferList|).
@@ -2200,7 +2216,7 @@ The following abstract operations operate on {{ReadableStream}} instances at a h 1. Otherwise: 1. Let |value| be ? [$IteratorValue$](|iterResult|). 1. Perform ! [$ReadableStreamDefaultControllerEnqueue$](|stream|.[=ReadableStream/[[controller]]=], - |value|). + |value|, « »). 1. Let |cancelAlgorithm| be the following steps, given |reason|: @@ -2362,8 +2378,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. @@ -2374,6 +2390,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|).

@@ -2414,10 +2432,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|, « »). 1. If |canceled2| is false, perform ! [$ReadableStreamDefaultControllerEnqueue$](|branch2|.[=ReadableStream/[[controller]]=], - |chunk2|). + |chunk2|, « »). 1. Set |reading| to false. 1. If |readAgain| is true, perform |pullAlgorithm|. @@ -3077,13 +3095,21 @@ The following abstract operations support the implementation of the
ReadableStreamDefaultControllerEnqueue(|controller|, - |chunk|) performs the following steps: + |chunk|, |transferList|) performs the following steps: + 1. Assert: |transferList| [=list/is empty=] or |controller|.[=ReadableStreamDefaultController/[[isOwning]]=] is true. 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. Let |result| be [$StructuredTransferOrClone$](|chunk|, |transferList|). + 1. If |result| is an abrupt completion, + 1. Perform ! [$ReadableStreamDefaultControllerError$](|controller|, |result|.\[[Value]]). + 1. Return |result|. + 1. Set |internalChunk| to |result|.\[[Value]]. + 1. Perform ! [$ReadableStreamFulfillReadRequest$](|stream|, |internalChunk|, false). 1. Otherwise, 1. Let |result| be the result of performing |controller|.[=ReadableStreamDefaultController/[[strategySizeAlgorithm]]=], passing in |chunk|, @@ -3092,7 +3118,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|. @@ -3156,7 +3182,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|. @@ -3166,8 +3192,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|. @@ -3192,6 +3219,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 @@ -3205,7 +3234,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

@@ -5351,7 +5380,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, « »). 1. Perform ! [$WritableStreamDefaultControllerAdvanceQueueIfNeeded$](|controller|). @@ -5455,7 +5484,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|, « »). 1. If |enqueueResult| is an abrupt completion, 1. Perform ! [$WritableStreamDefaultControllerErrorIfNeeded$](|controller|, |enqueueResult|.\[[Value]]). @@ -6095,7 +6124,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|, « »). 1. If |enqueueResult| is an abrupt completion, 1. Perform ! [$TransformStreamErrorWritableAndUnblockWrite$](|stream|, |enqueueResult|.\[[Value]]). @@ -6626,13 +6655,16 @@ 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 an \[[isOwning]] internal slot whose value is true, perform the following steps: + 1. Set |enqueuedValue| to ? [$StructuredTransferOrClone$](|value|, |transferList|). + 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|.
@@ -6652,6 +6684,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 an \[[isOwning]] internal slot whose value is true, perform the following steps until |container|.\[[queue]] + [=list/is empty=]: + 1. Let |chunk| be ! [$DequeueValue$]([=this=]). + 1. If |chunk| has [=dispose steps=], perform the [=dispose steps=] given |chunk|. 1. Set |container|.\[[queue]] to a new empty [=list=]. 1. Set |container|.\[[queueTotalSize]] to 0. @@ -6711,7 +6747,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|, « »). 1. Otherwise, if |type| is "`close`", 1. Perform ! [$ReadableStreamDefaultControllerClose$](|controller|). 1. Disentangle |port|. @@ -6860,6 +6896,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 @@ -6992,13 +7036,14 @@ 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. If |transferList| is null, set |transferList| to an [=list/is empty|empty=] list. 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 5123c8c0d..286f0e28b 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..8abc763d0 100644 --- a/reference-implementation/lib/ReadableStreamDefaultController-impl.js +++ b/reference-implementation/lib/ReadableStreamDefaultController-impl.js @@ -17,12 +17,16 @@ exports.implementation = class ReadableStreamDefaultControllerImpl { aos.ReadableStreamDefaultControllerClose(this); } - enqueue(chunk) { + enqueue(chunk, options) { + const transferList = options.transfer; + if (transferList.length && !this._isOwning) { + throw new TypeError('The stream is not an owning stream and cannot make use of options'); + } 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..041c731c5 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..ed2834374 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; }; @@ -37,9 +39,23 @@ exports.PeekQueueValue = container => { return pair.value; }; +const disposeStepsSymbol = Symbol('dispose-steps'); + 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[disposeStepsSymbol] === 'function') { + try { + value[disposeStepsSymbol](); + } 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 5cd4814b3..b40fd5a55 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, Call, CopyDataBlockBytes, CreateArrayFromList, GetIterator, GetMethod, IsDetachedBuffer, IteratorComplete, IteratorNext, IteratorValue, TransferArrayBuffer, typeIsObject } = 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, @@ -90,7 +90,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; @@ -341,7 +341,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) { @@ -393,10 +393,10 @@ function ReadableStreamDefaultTee(stream, cloneForBranch2) { // } if (canceled1 === false) { - ReadableStreamDefaultControllerEnqueue(branch1._controller, chunk1); + ReadableStreamDefaultControllerEnqueue(branch1._controller, chunk1, []); } if (canceled2 === false) { - ReadableStreamDefaultControllerEnqueue(branch2._controller, chunk2); + ReadableStreamDefaultControllerEnqueue(branch2._controller, chunk2, []); } reading = false; @@ -1075,7 +1075,7 @@ function ReadableStreamDefaultControllerClose(controller) { } } -function ReadableStreamDefaultControllerEnqueue(controller, chunk) { +function ReadableStreamDefaultControllerEnqueue(controller, chunk, transferList) { if (ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) === false) { return; } @@ -1083,6 +1083,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; @@ -1094,7 +1102,7 @@ function ReadableStreamDefaultControllerEnqueue(controller, chunk) { } try { - EnqueueValueWithSize(controller, chunk, chunkSize); + EnqueueValueWithSize(controller, chunk, chunkSize, transferList); } catch (enqueueE) { ReadableStreamDefaultControllerError(controller, enqueueE); throw enqueueE; @@ -1149,7 +1157,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; @@ -1170,6 +1178,8 @@ function SetUpReadableStreamDefaultController( controller._pullAlgorithm = pullAlgorithm; controller._cancelAlgorithm = cancelAlgorithm; + controller._isOwning = isOwning; + stream._controller = controller; const startResult = startAlgorithm(); @@ -1196,7 +1206,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); } @@ -1208,8 +1218,8 @@ function SetUpReadableStreamDefaultControllerFromUnderlyingSource( } SetUpReadableStreamDefaultController( - stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm - ); + stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, highWaterMark, sizeAlgorithm, + isOwning); } // Byte stream controllers @@ -1911,7 +1921,7 @@ function ReadableStreamFromIterable(asyncIterable) { ReadableStreamDefaultControllerClose(stream._controller); } else { const value = IteratorValue(iterResult); - ReadableStreamDefaultControllerEnqueue(stream._controller, value); + ReadableStreamDefaultControllerEnqueue(stream._controller, value, []); } }); } diff --git a/reference-implementation/lib/abstract-ops/transform-streams.js b/reference-implementation/lib/abstract-ops/transform-streams.js index 4edc5f10b..d78fdbc45 100644 --- a/reference-implementation/lib/abstract-ops/transform-streams.js +++ b/reference-implementation/lib/abstract-ops/transform-streams.js @@ -165,7 +165,7 @@ function TransformStreamDefaultControllerEnqueue(controller, chunk) { // accept TransformStreamDefaultControllerEnqueue() calls. try { - ReadableStreamDefaultControllerEnqueue(readableController, chunk); + ReadableStreamDefaultControllerEnqueue(readableController, chunk, []); } 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 0f685159e..5cc9021af 100644 --- a/reference-implementation/run-web-platform-tests.js +++ b/reference-implementation/run-web-platform-tests.js @@ -39,7 +39,6 @@ async function main() { 'readable-byte-streams/non-transferable-buffers.any.html', 'readable-streams/owning-type-message-port.any.html', // disabled due to MessagePort use. 'readable-streams/owning-type-video-frame.any.html', // disabled due to VideoFrame use. - 'readable-streams/owning-type.any.html', // FIXME: reenable this test once owning type PR lands. 'transferable/transform-stream-members.any.html' // FIXME: reenable if structuredClone is aligned. ]; const anyTestPattern = /\.any\.html$/; @@ -64,6 +63,7 @@ async function main() { } }; }; + window.structuredClone = globalThis.structuredClone; window.eval(bundledJS); }, filter(testPath) {