summaryrefslogtreecommitdiff
path: root/cli/js2/11_streams.js
diff options
context:
space:
mode:
Diffstat (limited to 'cli/js2/11_streams.js')
-rw-r--r--cli/js2/11_streams.js3290
1 files changed, 0 insertions, 3290 deletions
diff --git a/cli/js2/11_streams.js b/cli/js2/11_streams.js
deleted file mode 100644
index 4bdbfbc5c..000000000
--- a/cli/js2/11_streams.js
+++ /dev/null
@@ -1,3290 +0,0 @@
-// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
-
-// This code closely follows the WHATWG Stream Specification
-// See: https://streams.spec.whatwg.org/
-//
-// There are some parts that are not fully implemented, and there are some
-// comments which point to steps of the specification that are not implemented.
-
-((window) => {
- /* eslint-disable @typescript-eslint/no-explicit-any,require-await */
-
- const { AbortSignal } = window.__bootstrap.abortSignal;
- const { cloneValue, setFunctionName } = window.__bootstrap.webUtil;
- const { assert, AssertionError } = window.__bootstrap.util;
- const { customInspect, inspect } = window.__bootstrap.console;
-
- const sym = {
- abortAlgorithm: Symbol("abortAlgorithm"),
- abortSteps: Symbol("abortSteps"),
- asyncIteratorReader: Symbol("asyncIteratorReader"),
- autoAllocateChunkSize: Symbol("autoAllocateChunkSize"),
- backpressure: Symbol("backpressure"),
- backpressureChangePromise: Symbol("backpressureChangePromise"),
- byobRequest: Symbol("byobRequest"),
- cancelAlgorithm: Symbol("cancelAlgorithm"),
- cancelSteps: Symbol("cancelSteps"),
- closeAlgorithm: Symbol("closeAlgorithm"),
- closedPromise: Symbol("closedPromise"),
- closeRequest: Symbol("closeRequest"),
- closeRequested: Symbol("closeRequested"),
- controlledReadableByteStream: Symbol(
- "controlledReadableByteStream",
- ),
- controlledReadableStream: Symbol("controlledReadableStream"),
- controlledTransformStream: Symbol("controlledTransformStream"),
- controlledWritableStream: Symbol("controlledWritableStream"),
- disturbed: Symbol("disturbed"),
- errorSteps: Symbol("errorSteps"),
- flushAlgorithm: Symbol("flushAlgorithm"),
- forAuthorCode: Symbol("forAuthorCode"),
- inFlightWriteRequest: Symbol("inFlightWriteRequest"),
- inFlightCloseRequest: Symbol("inFlightCloseRequest"),
- isFakeDetached: Symbol("isFakeDetached"),
- ownerReadableStream: Symbol("ownerReadableStream"),
- ownerWritableStream: Symbol("ownerWritableStream"),
- pendingAbortRequest: Symbol("pendingAbortRequest"),
- preventCancel: Symbol("preventCancel"),
- pullAgain: Symbol("pullAgain"),
- pullAlgorithm: Symbol("pullAlgorithm"),
- pulling: Symbol("pulling"),
- pullSteps: Symbol("pullSteps"),
- queue: Symbol("queue"),
- queueTotalSize: Symbol("queueTotalSize"),
- readable: Symbol("readable"),
- readableStreamController: Symbol("readableStreamController"),
- reader: Symbol("reader"),
- readRequests: Symbol("readRequests"),
- readyPromise: Symbol("readyPromise"),
- started: Symbol("started"),
- state: Symbol("state"),
- storedError: Symbol("storedError"),
- strategyHWM: Symbol("strategyHWM"),
- strategySizeAlgorithm: Symbol("strategySizeAlgorithm"),
- transformAlgorithm: Symbol("transformAlgorithm"),
- transformStreamController: Symbol("transformStreamController"),
- writableStreamController: Symbol("writableStreamController"),
- writeAlgorithm: Symbol("writeAlgorithm"),
- writable: Symbol("writable"),
- writer: Symbol("writer"),
- writeRequests: Symbol("writeRequests"),
- };
- class ReadableByteStreamController {
- constructor() {
- throw new TypeError(
- "ReadableByteStreamController's constructor cannot be called.",
- );
- }
-
- get byobRequest() {
- return undefined;
- }
-
- get desiredSize() {
- if (!isReadableByteStreamController(this)) {
- throw new TypeError("Invalid ReadableByteStreamController.");
- }
- return readableByteStreamControllerGetDesiredSize(this);
- }
-
- close() {
- if (!isReadableByteStreamController(this)) {
- throw new TypeError("Invalid ReadableByteStreamController.");
- }
- if (this[sym.closeRequested]) {
- throw new TypeError("Closed already requested.");
- }
- if (this[sym.controlledReadableByteStream][sym.state] !== "readable") {
- throw new TypeError(
- "ReadableByteStreamController's stream is not in a readable state.",
- );
- }
- readableByteStreamControllerClose(this);
- }
-
- enqueue(chunk) {
- if (!isReadableByteStreamController(this)) {
- throw new TypeError("Invalid ReadableByteStreamController.");
- }
- if (this[sym.closeRequested]) {
- throw new TypeError("Closed already requested.");
- }
- if (this[sym.controlledReadableByteStream][sym.state] !== "readable") {
- throw new TypeError(
- "ReadableByteStreamController's stream is not in a readable state.",
- );
- }
- if (!ArrayBuffer.isView(chunk)) {
- throw new TypeError(
- "You can only enqueue array buffer views when using a ReadableByteStreamController",
- );
- }
- if (isDetachedBuffer(chunk.buffer)) {
- throw new TypeError(
- "Cannot enqueue a view onto a detached ArrayBuffer",
- );
- }
- readableByteStreamControllerEnqueue(this, chunk);
- }
-
- error(error) {
- if (!isReadableByteStreamController(this)) {
- throw new TypeError("Invalid ReadableByteStreamController.");
- }
- readableByteStreamControllerError(this, error);
- }
-
- [sym.cancelSteps](reason) {
- // 3.11.5.1.1 If this.[[pendingPullIntos]] is not empty,
- resetQueue(this);
- const result = this[sym.cancelAlgorithm](reason);
- readableByteStreamControllerClearAlgorithms(this);
- return result;
- }
-
- [sym.pullSteps]() {
- const stream = this[sym.controlledReadableByteStream];
- assert(readableStreamHasDefaultReader(stream));
- if (this[sym.queueTotalSize] > 0) {
- assert(readableStreamGetNumReadRequests(stream) === 0);
- const entry = this[sym.queue].shift();
- assert(entry);
- this[sym.queueTotalSize] -= entry.size;
- readableByteStreamControllerHandleQueueDrain(this);
- const view = new Uint8Array(entry.value, entry.offset, entry.size);
- return Promise.resolve(
- readableStreamCreateReadResult(
- view,
- false,
- stream[sym.reader][sym.forAuthorCode],
- ),
- );
- }
- // 3.11.5.2.5 If autoAllocateChunkSize is not undefined,
- const promise = readableStreamAddReadRequest(stream);
- readableByteStreamControllerCallPullIfNeeded(this);
- return promise;
- }
-
- [customInspect]() {
- return `${this.constructor.name} { byobRequest: ${
- String(this.byobRequest)
- }, desiredSize: ${String(this.desiredSize)} }`;
- }
- }
-
- class ReadableStreamDefaultController {
- constructor() {
- throw new TypeError(
- "ReadableStreamDefaultController's constructor cannot be called.",
- );
- }
-
- get desiredSize() {
- if (!isReadableStreamDefaultController(this)) {
- throw new TypeError("Invalid ReadableStreamDefaultController.");
- }
- return readableStreamDefaultControllerGetDesiredSize(this);
- }
-
- close() {
- if (!isReadableStreamDefaultController(this)) {
- throw new TypeError("Invalid ReadableStreamDefaultController.");
- }
- if (!readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
- throw new TypeError(
- "ReadableStreamDefaultController cannot close or enqueue.",
- );
- }
- readableStreamDefaultControllerClose(this);
- }
-
- enqueue(chunk) {
- if (!isReadableStreamDefaultController(this)) {
- throw new TypeError("Invalid ReadableStreamDefaultController.");
- }
- if (!readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
- throw new TypeError("ReadableSteamController cannot enqueue.");
- }
- return readableStreamDefaultControllerEnqueue(this, chunk);
- }
-
- error(error) {
- if (!isReadableStreamDefaultController(this)) {
- throw new TypeError("Invalid ReadableStreamDefaultController.");
- }
- readableStreamDefaultControllerError(this, error);
- }
-
- [sym.cancelSteps](reason) {
- resetQueue(this);
- const result = this[sym.cancelAlgorithm](reason);
- readableStreamDefaultControllerClearAlgorithms(this);
- return result;
- }
-
- [sym.pullSteps]() {
- const stream = this[sym.controlledReadableStream];
- if (this[sym.queue].length) {
- const chunk = dequeueValue(this);
- if (this[sym.closeRequested] && this[sym.queue].length === 0) {
- readableStreamDefaultControllerClearAlgorithms(this);
- readableStreamClose(stream);
- } else {
- readableStreamDefaultControllerCallPullIfNeeded(this);
- }
- return Promise.resolve(
- readableStreamCreateReadResult(
- chunk,
- false,
- stream[sym.reader][sym.forAuthorCode],
- ),
- );
- }
- const pendingPromise = readableStreamAddReadRequest(stream);
- readableStreamDefaultControllerCallPullIfNeeded(this);
- return pendingPromise;
- }
-
- [customInspect]() {
- return `${this.constructor.name} { desiredSize: ${
- String(this.desiredSize)
- } }`;
- }
- }
-
- class ReadableStreamDefaultReader {
- constructor(stream) {
- if (!isReadableStream(stream)) {
- throw new TypeError("stream is not a ReadableStream.");
- }
- if (isReadableStreamLocked(stream)) {
- throw new TypeError("stream is locked.");
- }
- readableStreamReaderGenericInitialize(this, stream);
- this[sym.readRequests] = [];
- }
-
- get closed() {
- if (!isReadableStreamDefaultReader(this)) {
- return Promise.reject(
- new TypeError("Invalid ReadableStreamDefaultReader."),
- );
- }
- return (
- this[sym.closedPromise].promise ??
- Promise.reject(new TypeError("Invalid reader."))
- );
- }
-
- cancel(reason) {
- if (!isReadableStreamDefaultReader(this)) {
- return Promise.reject(
- new TypeError("Invalid ReadableStreamDefaultReader."),
- );
- }
- if (!this[sym.ownerReadableStream]) {
- return Promise.reject(new TypeError("Invalid reader."));
- }
- return readableStreamReaderGenericCancel(this, reason);
- }
-
- read() {
- if (!isReadableStreamDefaultReader(this)) {
- return Promise.reject(
- new TypeError("Invalid ReadableStreamDefaultReader."),
- );
- }
- if (!this[sym.ownerReadableStream]) {
- return Promise.reject(new TypeError("Invalid reader."));
- }
- return readableStreamDefaultReaderRead(this);
- }
-
- releaseLock() {
- if (!isReadableStreamDefaultReader(this)) {
- throw new TypeError("Invalid ReadableStreamDefaultReader.");
- }
- if (this[sym.ownerReadableStream] === undefined) {
- return;
- }
- if (this[sym.readRequests].length) {
- throw new TypeError("Cannot release lock with pending read requests.");
- }
- readableStreamReaderGenericRelease(this);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { closed: Promise }`;
- }
- }
-
- const AsyncIteratorPrototype = Object
- .getPrototypeOf(Object.getPrototypeOf(async function* () {}).prototype);
-
- const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
- next() {
- if (!isReadableStreamAsyncIterator(this)) {
- return Promise.reject(
- new TypeError("invalid ReadableStreamAsyncIterator."),
- );
- }
- const reader = this[sym.asyncIteratorReader];
- if (!reader[sym.ownerReadableStream]) {
- return Promise.reject(
- new TypeError("reader owner ReadableStream is undefined."),
- );
- }
- return readableStreamDefaultReaderRead(reader).then((result) => {
- assert(typeof result === "object");
- const { done } = result;
- assert(typeof done === "boolean");
- if (done) {
- readableStreamReaderGenericRelease(reader);
- }
- const { value } = result;
- return readableStreamCreateReadResult(value, done, true);
- });
- },
- return(
- value,
- ) {
- if (!isReadableStreamAsyncIterator(this)) {
- return Promise.reject(
- new TypeError("invalid ReadableStreamAsyncIterator."),
- );
- }
- const reader = this[sym.asyncIteratorReader];
- if (!reader[sym.ownerReadableStream]) {
- return Promise.reject(
- new TypeError("reader owner ReadableStream is undefined."),
- );
- }
- if (reader[sym.readRequests].length) {
- return Promise.reject(
- new TypeError("reader has outstanding read requests."),
- );
- }
- if (!this[sym.preventCancel]) {
- const result = readableStreamReaderGenericCancel(reader, value);
- readableStreamReaderGenericRelease(reader);
- return result.then(() =>
- readableStreamCreateReadResult(value, true, true)
- );
- }
- readableStreamReaderGenericRelease(reader);
- return Promise.resolve(
- readableStreamCreateReadResult(value, true, true),
- );
- },
- }, AsyncIteratorPrototype);
-
- class ReadableStream {
- constructor(
- underlyingSource = {},
- strategy = {},
- ) {
- initializeReadableStream(this);
- const { size } = strategy;
- let { highWaterMark } = strategy;
- const { type } = underlyingSource;
-
- if (isUnderlyingByteSource(underlyingSource)) {
- if (size !== undefined) {
- throw new RangeError(
- `When underlying source is "bytes", strategy.size must be undefined.`,
- );
- }
- highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark ?? 0);
- setUpReadableByteStreamControllerFromUnderlyingSource(
- this,
- underlyingSource,
- highWaterMark,
- );
- } else if (type === undefined) {
- const sizeAlgorithm = makeSizeAlgorithmFromSizeFunction(size);
- highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark ?? 1);
- setUpReadableStreamDefaultControllerFromUnderlyingSource(
- this,
- underlyingSource,
- highWaterMark,
- sizeAlgorithm,
- );
- } else {
- throw new RangeError(
- `Valid values for underlyingSource are "bytes" or undefined. Received: "${type}".`,
- );
- }
- }
-
- get locked() {
- if (!isReadableStream(this)) {
- throw new TypeError("Invalid ReadableStream.");
- }
- return isReadableStreamLocked(this);
- }
-
- cancel(reason) {
- if (!isReadableStream(this)) {
- return Promise.reject(new TypeError("Invalid ReadableStream."));
- }
- if (isReadableStreamLocked(this)) {
- return Promise.reject(
- new TypeError("Cannot cancel a locked ReadableStream."),
- );
- }
- return readableStreamCancel(this, reason);
- }
-
- getIterator({
- preventCancel,
- } = {}) {
- if (!isReadableStream(this)) {
- throw new TypeError("Invalid ReadableStream.");
- }
- const reader = acquireReadableStreamDefaultReader(this);
- const iterator = Object.create(ReadableStreamAsyncIteratorPrototype);
- iterator[sym.asyncIteratorReader] = reader;
- iterator[sym.preventCancel] = Boolean(preventCancel);
- return iterator;
- }
-
- getReader({ mode } = {}) {
- if (!isReadableStream(this)) {
- throw new TypeError("Invalid ReadableStream.");
- }
- if (mode === undefined) {
- return acquireReadableStreamDefaultReader(this, true);
- }
- mode = String(mode);
- // 3.2.5.4.4 If mode is "byob", return ? AcquireReadableStreamBYOBReader(this, true).
- throw new RangeError(`Unsupported mode "${mode}"`);
- }
-
- pipeThrough(
- {
- writable,
- readable,
- },
- { preventClose, preventAbort, preventCancel, signal } = {},
- ) {
- if (!isReadableStream(this)) {
- throw new TypeError("Invalid ReadableStream.");
- }
- if (!isWritableStream(writable)) {
- throw new TypeError("writable is not a valid WritableStream.");
- }
- if (!isReadableStream(readable)) {
- throw new TypeError("readable is not a valid ReadableStream.");
- }
- preventClose = Boolean(preventClose);
- preventAbort = Boolean(preventAbort);
- preventCancel = Boolean(preventCancel);
- if (signal && !(signal instanceof AbortSignal)) {
- throw new TypeError("Invalid signal.");
- }
- if (isReadableStreamLocked(this)) {
- throw new TypeError("ReadableStream is locked.");
- }
- if (isWritableStreamLocked(writable)) {
- throw new TypeError("writable is locked.");
- }
- const promise = readableStreamPipeTo(
- this,
- writable,
- preventClose,
- preventAbort,
- preventCancel,
- signal,
- );
- setPromiseIsHandledToTrue(promise);
- return readable;
- }
-
- pipeTo(
- dest,
- { preventClose, preventAbort, preventCancel, signal } = {},
- ) {
- if (!isReadableStream(this)) {
- return Promise.reject(new TypeError("Invalid ReadableStream."));
- }
- if (!isWritableStream(dest)) {
- return Promise.reject(
- new TypeError("dest is not a valid WritableStream."),
- );
- }
- preventClose = Boolean(preventClose);
- preventAbort = Boolean(preventAbort);
- preventCancel = Boolean(preventCancel);
- if (signal && !(signal instanceof AbortSignal)) {
- return Promise.reject(new TypeError("Invalid signal."));
- }
- if (isReadableStreamLocked(this)) {
- return Promise.reject(new TypeError("ReadableStream is locked."));
- }
- if (isWritableStreamLocked(dest)) {
- return Promise.reject(new TypeError("dest is locked."));
- }
- return readableStreamPipeTo(
- this,
- dest,
- preventClose,
- preventAbort,
- preventCancel,
- signal,
- );
- }
-
- tee() {
- if (!isReadableStream(this)) {
- throw new TypeError("Invalid ReadableStream.");
- }
- return readableStreamTee(this, false);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { locked: ${String(this.locked)} }`;
- }
-
- [Symbol.asyncIterator](
- options = {},
- ) {
- return this.getIterator(options);
- }
- }
-
- class TransformStream {
- constructor(
- transformer = {},
- writableStrategy = {},
- readableStrategy = {},
- ) {
- const writableSizeFunction = writableStrategy.size;
- let writableHighWaterMark = writableStrategy.highWaterMark;
- const readableSizeFunction = readableStrategy.size;
- let readableHighWaterMark = readableStrategy.highWaterMark;
- const writableType = transformer.writableType;
- if (writableType !== undefined) {
- throw new RangeError(
- `Expected transformer writableType to be undefined, received "${
- String(writableType)
- }"`,
- );
- }
- const writableSizeAlgorithm = makeSizeAlgorithmFromSizeFunction(
- writableSizeFunction,
- );
- if (writableHighWaterMark === undefined) {
- writableHighWaterMark = 1;
- }
- writableHighWaterMark = validateAndNormalizeHighWaterMark(
- writableHighWaterMark,
- );
- const readableType = transformer.readableType;
- if (readableType !== undefined) {
- throw new RangeError(
- `Expected transformer readableType to be undefined, received "${
- String(readableType)
- }"`,
- );
- }
- const readableSizeAlgorithm = makeSizeAlgorithmFromSizeFunction(
- readableSizeFunction,
- );
- if (readableHighWaterMark === undefined) {
- readableHighWaterMark = 1;
- }
- readableHighWaterMark = validateAndNormalizeHighWaterMark(
- readableHighWaterMark,
- );
- const startPromise = getDeferred();
- initializeTransformStream(
- this,
- startPromise.promise,
- writableHighWaterMark,
- writableSizeAlgorithm,
- readableHighWaterMark,
- readableSizeAlgorithm,
- );
- // the brand check expects this, and the brand check occurs in the following
- // but the property hasn't been defined.
- Object.defineProperty(this, sym.transformStreamController, {
- value: undefined,
- writable: true,
- configurable: true,
- });
- setUpTransformStreamDefaultControllerFromTransformer(this, transformer);
- const startResult = invokeOrNoop(
- transformer,
- "start",
- this[sym.transformStreamController],
- );
- startPromise.resolve(startResult);
- }
-
- get readable() {
- if (!isTransformStream(this)) {
- throw new TypeError("Invalid TransformStream.");
- }
- return this[sym.readable];
- }
-
- get writable() {
- if (!isTransformStream(this)) {
- throw new TypeError("Invalid TransformStream.");
- }
- return this[sym.writable];
- }
-
- [customInspect]() {
- return `${this.constructor.name} {\n readable: ${
- inspect(this.readable)
- }\n writable: ${inspect(this.writable)}\n}`;
- }
- }
-
- class TransformStreamDefaultController {
- constructor() {
- throw new TypeError(
- "TransformStreamDefaultController's constructor cannot be called.",
- );
- }
-
- get desiredSize() {
- if (!isTransformStreamDefaultController(this)) {
- throw new TypeError("Invalid TransformStreamDefaultController.");
- }
- const readableController = this[sym.controlledTransformStream][
- sym.readable
- ][sym.readableStreamController];
- return readableStreamDefaultControllerGetDesiredSize(
- readableController,
- );
- }
-
- enqueue(chunk) {
- if (!isTransformStreamDefaultController(this)) {
- throw new TypeError("Invalid TransformStreamDefaultController.");
- }
- transformStreamDefaultControllerEnqueue(this, chunk);
- }
-
- error(reason) {
- if (!isTransformStreamDefaultController(this)) {
- throw new TypeError("Invalid TransformStreamDefaultController.");
- }
- transformStreamDefaultControllerError(this, reason);
- }
-
- terminate() {
- if (!isTransformStreamDefaultController(this)) {
- throw new TypeError("Invalid TransformStreamDefaultController.");
- }
- transformStreamDefaultControllerTerminate(this);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { desiredSize: ${
- String(this.desiredSize)
- } }`;
- }
- }
-
- class WritableStreamDefaultController {
- constructor() {
- throw new TypeError(
- "WritableStreamDefaultController's constructor cannot be called.",
- );
- }
-
- error(e) {
- if (!isWritableStreamDefaultController(this)) {
- throw new TypeError("Invalid WritableStreamDefaultController.");
- }
- const state = this[sym.controlledWritableStream][sym.state];
- if (state !== "writable") {
- return;
- }
- writableStreamDefaultControllerError(this, e);
- }
-
- [sym.abortSteps](reason) {
- const result = this[sym.abortAlgorithm](reason);
- writableStreamDefaultControllerClearAlgorithms(this);
- return result;
- }
-
- [sym.errorSteps]() {
- resetQueue(this);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { }`;
- }
- }
-
- class WritableStreamDefaultWriter {
- constructor(stream) {
- if (!isWritableStream(stream)) {
- throw new TypeError("Invalid stream.");
- }
- if (isWritableStreamLocked(stream)) {
- throw new TypeError("Cannot create a writer for a locked stream.");
- }
- this[sym.ownerWritableStream] = stream;
- stream[sym.writer] = this;
- const state = stream[sym.state];
- if (state === "writable") {
- if (
- !writableStreamCloseQueuedOrInFlight(stream) &&
- stream[sym.backpressure]
- ) {
- this[sym.readyPromise] = getDeferred();
- } else {
- this[sym.readyPromise] = { promise: Promise.resolve() };
- }
- this[sym.closedPromise] = getDeferred();
- } else if (state === "erroring") {
- this[sym.readyPromise] = {
- promise: Promise.reject(stream[sym.storedError]),
- };
- setPromiseIsHandledToTrue(this[sym.readyPromise].promise);
- this[sym.closedPromise] = getDeferred();
- } else if (state === "closed") {
- this[sym.readyPromise] = { promise: Promise.resolve() };
- this[sym.closedPromise] = { promise: Promise.resolve() };
- } else {
- assert(state === "errored");
- const storedError = stream[sym.storedError];
- this[sym.readyPromise] = { promise: Promise.reject(storedError) };
- setPromiseIsHandledToTrue(this[sym.readyPromise].promise);
- this[sym.closedPromise] = { promise: Promise.reject(storedError) };
- setPromiseIsHandledToTrue(this[sym.closedPromise].promise);
- }
- }
-
- get closed() {
- if (!isWritableStreamDefaultWriter(this)) {
- return Promise.reject(
- new TypeError("Invalid WritableStreamDefaultWriter."),
- );
- }
- return this[sym.closedPromise].promise;
- }
-
- get desiredSize() {
- if (!isWritableStreamDefaultWriter(this)) {
- throw new TypeError("Invalid WritableStreamDefaultWriter.");
- }
- if (!this[sym.ownerWritableStream]) {
- throw new TypeError("WritableStreamDefaultWriter has no owner.");
- }
- return writableStreamDefaultWriterGetDesiredSize(this);
- }
-
- get ready() {
- if (!isWritableStreamDefaultWriter(this)) {
- return Promise.reject(
- new TypeError("Invalid WritableStreamDefaultWriter."),
- );
- }
- return this[sym.readyPromise].promise;
- }
-
- abort(reason) {
- if (!isWritableStreamDefaultWriter(this)) {
- return Promise.reject(
- new TypeError("Invalid WritableStreamDefaultWriter."),
- );
- }
- if (!this[sym.ownerWritableStream]) {
- Promise.reject(
- new TypeError("WritableStreamDefaultWriter has no owner."),
- );
- }
- return writableStreamDefaultWriterAbort(this, reason);
- }
-
- close() {
- if (!isWritableStreamDefaultWriter(this)) {
- return Promise.reject(
- new TypeError("Invalid WritableStreamDefaultWriter."),
- );
- }
- const stream = this[sym.ownerWritableStream];
- if (!stream) {
- Promise.reject(
- new TypeError("WritableStreamDefaultWriter has no owner."),
- );
- }
- if (writableStreamCloseQueuedOrInFlight(stream)) {
- Promise.reject(
- new TypeError("Stream is in an invalid state to be closed."),
- );
- }
- return writableStreamDefaultWriterClose(this);
- }
-
- releaseLock() {
- if (!isWritableStreamDefaultWriter(this)) {
- throw new TypeError("Invalid WritableStreamDefaultWriter.");
- }
- const stream = this[sym.ownerWritableStream];
- if (!stream) {
- return;
- }
- assert(stream[sym.writer]);
- writableStreamDefaultWriterRelease(this);
- }
-
- write(chunk) {
- if (!isWritableStreamDefaultWriter(this)) {
- return Promise.reject(
- new TypeError("Invalid WritableStreamDefaultWriter."),
- );
- }
- if (!this[sym.ownerWritableStream]) {
- Promise.reject(
- new TypeError("WritableStreamDefaultWriter has no owner."),
- );
- }
- return writableStreamDefaultWriterWrite(this, chunk);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { closed: Promise, desiredSize: ${
- String(this.desiredSize)
- }, ready: Promise }`;
- }
- }
-
- class WritableStream {
- constructor(
- underlyingSink = {},
- strategy = {},
- ) {
- initializeWritableStream(this);
- const size = strategy.size;
- let highWaterMark = strategy.highWaterMark ?? 1;
- const { type } = underlyingSink;
- if (type !== undefined) {
- throw new RangeError(`Sink type of "${String(type)}" not supported.`);
- }
- const sizeAlgorithm = makeSizeAlgorithmFromSizeFunction(size);
- highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark);
- setUpWritableStreamDefaultControllerFromUnderlyingSink(
- this,
- underlyingSink,
- highWaterMark,
- sizeAlgorithm,
- );
- }
-
- get locked() {
- if (!isWritableStream(this)) {
- throw new TypeError("Invalid WritableStream.");
- }
- return isWritableStreamLocked(this);
- }
-
- abort(reason) {
- if (!isWritableStream(this)) {
- return Promise.reject(new TypeError("Invalid WritableStream."));
- }
- if (isWritableStreamLocked(this)) {
- return Promise.reject(
- new TypeError("Cannot abort a locked WritableStream."),
- );
- }
- return writableStreamAbort(this, reason);
- }
-
- close() {
- if (!isWritableStream(this)) {
- return Promise.reject(new TypeError("Invalid WritableStream."));
- }
- if (isWritableStreamLocked(this)) {
- return Promise.reject(
- new TypeError("Cannot abort a locked WritableStream."),
- );
- }
- if (writableStreamCloseQueuedOrInFlight(this)) {
- return Promise.reject(
- new TypeError("Cannot close an already closing WritableStream."),
- );
- }
- return writableStreamClose(this);
- }
-
- getWriter() {
- if (!isWritableStream(this)) {
- throw new TypeError("Invalid WritableStream.");
- }
- return acquireWritableStreamDefaultWriter(this);
- }
-
- [customInspect]() {
- return `${this.constructor.name} { locked: ${String(this.locked)} }`;
- }
- }
-
- function acquireReadableStreamDefaultReader(
- stream,
- forAuthorCode = false,
- ) {
- const reader = new ReadableStreamDefaultReader(stream);
- reader[sym.forAuthorCode] = forAuthorCode;
- return reader;
- }
-
- function acquireWritableStreamDefaultWriter(
- stream,
- ) {
- return new WritableStreamDefaultWriter(stream);
- }
-
- function call(
- fn,
- v,
- args,
- ) {
- return Function.prototype.apply.call(fn, v, args);
- }
-
- function createAlgorithmFromUnderlyingMethod(
- underlyingObject,
- methodName,
- algoArgCount,
- ...extraArgs
- ) {
- const method = underlyingObject[methodName];
- if (method) {
- if (!isCallable(method)) {
- throw new TypeError("method is not callable");
- }
- if (algoArgCount === 0) {
- return async () => call(method, underlyingObject, extraArgs);
- } else {
- return async (arg) => {
- const fullArgs = [arg, ...extraArgs];
- return call(method, underlyingObject, fullArgs);
- };
- }
- }
- return async () => undefined;
- }
-
- function createReadableStream(
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark = 1,
- sizeAlgorithm = () => 1,
- ) {
- highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark);
- const stream = Object.create(
- ReadableStream.prototype,
- );
- initializeReadableStream(stream);
- const controller = Object.create(
- ReadableStreamDefaultController.prototype,
- );
- setUpReadableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- );
- return stream;
- }
-
- function createWritableStream(
- startAlgorithm,
- writeAlgorithm,
- closeAlgorithm,
- abortAlgorithm,
- highWaterMark = 1,
- sizeAlgorithm = () => 1,
- ) {
- highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark);
- const stream = Object.create(WritableStream.prototype);
- initializeWritableStream(stream);
- const controller = Object.create(
- WritableStreamDefaultController.prototype,
- );
- setUpWritableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- writeAlgorithm,
- closeAlgorithm,
- abortAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- );
- return stream;
- }
-
- function dequeueValue(container) {
- assert(sym.queue in container && sym.queueTotalSize in container);
- assert(container[sym.queue].length);
- const pair = container[sym.queue].shift();
- container[sym.queueTotalSize] -= pair.size;
- if (container[sym.queueTotalSize] <= 0) {
- container[sym.queueTotalSize] = 0;
- }
- return pair.value;
- }
-
- function enqueueValueWithSize(
- container,
- value,
- size,
- ) {
- assert(sym.queue in container && sym.queueTotalSize in container);
- size = Number(size);
- if (!isFiniteNonNegativeNumber(size)) {
- throw new RangeError("size must be a finite non-negative number.");
- }
- container[sym.queue].push({ value, size });
- container[sym.queueTotalSize] += size;
- }
-
- /** Non-spec mechanism to "unwrap" a promise and store it to be resolved
- * later. */
- function getDeferred() {
- let resolve;
- let reject;
- const promise = new Promise((res, rej) => {
- resolve = res;
- reject = rej;
- });
- return { promise, resolve: resolve, reject: reject };
- }
-
- function initializeReadableStream(
- stream,
- ) {
- stream[sym.state] = "readable";
- stream[sym.reader] = stream[sym.storedError] = undefined;
- stream[sym.disturbed] = false;
- }
-
- function initializeTransformStream(
- stream,
- startPromise,
- writableHighWaterMark,
- writableSizeAlgorithm,
- readableHighWaterMark,
- readableSizeAlgorithm,
- ) {
- const startAlgorithm = () => startPromise;
- const writeAlgorithm = (chunk) =>
- transformStreamDefaultSinkWriteAlgorithm(stream, chunk);
- const abortAlgorithm = (reason) =>
- transformStreamDefaultSinkAbortAlgorithm(stream, reason);
- const closeAlgorithm = () =>
- transformStreamDefaultSinkCloseAlgorithm(stream);
- stream[sym.writable] = createWritableStream(
- startAlgorithm,
- writeAlgorithm,
- closeAlgorithm,
- abortAlgorithm,
- writableHighWaterMark,
- writableSizeAlgorithm,
- );
- const pullAlgorithm = () =>
- transformStreamDefaultSourcePullAlgorithm(stream);
- const cancelAlgorithm = (reason) => {
- transformStreamErrorWritableAndUnblockWrite(stream, reason);
- return Promise.resolve(undefined);
- };
- stream[sym.readable] = createReadableStream(
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- readableHighWaterMark,
- readableSizeAlgorithm,
- );
- stream[sym.backpressure] = stream[sym.backpressureChangePromise] =
- undefined;
- transformStreamSetBackpressure(stream, true);
- Object.defineProperty(stream, sym.transformStreamController, {
- value: undefined,
- configurable: true,
- });
- }
-
- function initializeWritableStream(
- stream,
- ) {
- stream[sym.state] = "writable";
- stream[sym.storedError] = stream[sym.writer] = stream[
- sym.writableStreamController
- ] = stream[sym.inFlightWriteRequest] = stream[sym.closeRequest] = stream[
- sym.inFlightCloseRequest
- ] = stream[sym.pendingAbortRequest] = undefined;
- stream[sym.writeRequests] = [];
- stream[sym.backpressure] = false;
- }
-
- function invokeOrNoop(
- o,
- p,
- ...args
- ) {
- assert(o);
- const method = o[p];
- if (!method) {
- return undefined;
- }
- return call(method, o, args);
- }
-
- function isCallable(value) {
- return typeof value === "function";
- }
-
- function isDetachedBuffer(value) {
- return sym.isFakeDetached in value;
- }
-
- function isFiniteNonNegativeNumber(v) {
- return Number.isFinite(v) && (v) >= 0;
- }
-
- function isReadableByteStreamController(
- x,
- ) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.controlledReadableByteStream in x)
- );
- }
-
- function isReadableStream(x) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.readableStreamController in x)
- );
- }
-
- function isReadableStreamAsyncIterator(
- x,
- ) {
- if (typeof x !== "object" || x === null) {
- return false;
- }
- return sym.asyncIteratorReader in x;
- }
-
- function isReadableStreamDefaultController(
- x,
- ) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.controlledReadableStream in x)
- );
- }
-
- function isReadableStreamDefaultReader(
- x,
- ) {
- return !(typeof x !== "object" || x === null || !(sym.readRequests in x));
- }
-
- function isReadableStreamLocked(stream) {
- assert(isReadableStream(stream));
- return !!stream[sym.reader];
- }
-
- function isReadableStreamDisturbed(stream) {
- assert(isReadableStream(stream));
- return !!stream[sym.disturbed];
- }
-
- function isTransformStream(x) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.transformStreamController in x)
- );
- }
-
- function isTransformStreamDefaultController(
- x,
- ) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.controlledTransformStream in x)
- );
- }
-
- function isUnderlyingByteSource(
- underlyingSource,
- ) {
- const { type } = underlyingSource;
- const typeString = String(type);
- return typeString === "bytes";
- }
-
- function isWritableStream(x) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.writableStreamController in x)
- );
- }
-
- function isWritableStreamDefaultController(
- x,
- ) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.controlledWritableStream in x)
- );
- }
-
- function isWritableStreamDefaultWriter(
- x,
- ) {
- return !(
- typeof x !== "object" ||
- x === null ||
- !(sym.ownerWritableStream in x)
- );
- }
-
- function isWritableStreamLocked(stream) {
- assert(isWritableStream(stream));
- return stream[sym.writer] !== undefined;
- }
-
- function makeSizeAlgorithmFromSizeFunction(
- size,
- ) {
- if (size === undefined) {
- return () => 1;
- }
- if (typeof size !== "function") {
- throw new TypeError("size must be callable.");
- }
- return (chunk) => {
- return size.call(undefined, chunk);
- };
- }
-
- function peekQueueValue(container) {
- assert(sym.queue in container && sym.queueTotalSize in container);
- assert(container[sym.queue].length);
- const [pair] = container[sym.queue];
- return pair.value;
- }
-
- function readableByteStreamControllerShouldCallPull(
- controller,
- ) {
- const stream = controller[sym.controlledReadableByteStream];
- if (
- stream[sym.state] !== "readable" ||
- controller[sym.closeRequested] ||
- !controller[sym.started]
- ) {
- return false;
- }
- if (
- readableStreamHasDefaultReader(stream) &&
- readableStreamGetNumReadRequests(stream) > 0
- ) {
- return true;
- }
- // 3.13.25.6 If ! ReadableStreamHasBYOBReader(stream) is true and !
- // ReadableStreamGetNumReadIntoRequests(stream) > 0, return true.
- const desiredSize = readableByteStreamControllerGetDesiredSize(controller);
- assert(desiredSize !== null);
- return desiredSize > 0;
- }
-
- function readableByteStreamControllerCallPullIfNeeded(
- controller,
- ) {
- const shouldPull = readableByteStreamControllerShouldCallPull(controller);
- if (!shouldPull) {
- return;
- }
- if (controller[sym.pulling]) {
- controller[sym.pullAgain] = true;
- return;
- }
- assert(controller[sym.pullAgain] === false);
- controller[sym.pulling] = true;
- const pullPromise = controller[sym.pullAlgorithm]();
- setPromiseIsHandledToTrue(
- pullPromise.then(
- () => {
- controller[sym.pulling] = false;
- if (controller[sym.pullAgain]) {
- controller[sym.pullAgain] = false;
- readableByteStreamControllerCallPullIfNeeded(controller);
- }
- },
- (e) => {
- readableByteStreamControllerError(controller, e);
- },
- ),
- );
- }
-
- function readableByteStreamControllerClearAlgorithms(
- controller,
- ) {
- controller[sym.pullAlgorithm] = undefined;
- controller[sym.cancelAlgorithm] = undefined;
- }
-
- function readableByteStreamControllerClose(
- controller,
- ) {
- const stream = controller[sym.controlledReadableByteStream];
- if (controller[sym.closeRequested] || stream[sym.state] !== "readable") {
- return;
- }
- if (controller[sym.queueTotalSize] > 0) {
- controller[sym.closeRequested] = true;
- return;
- }
- // 3.13.6.4 If controller.[[pendingPullIntos]] is not empty, (BYOB Support)
- readableByteStreamControllerClearAlgorithms(controller);
- readableStreamClose(stream);
- }
-
- function readableByteStreamControllerEnqueue(
- controller,
- chunk,
- ) {
- const stream = controller[sym.controlledReadableByteStream];
- if (controller[sym.closeRequested] || stream[sym.state] !== "readable") {
- return;
- }
- const { buffer, byteOffset, byteLength } = chunk;
- const transferredBuffer = transferArrayBuffer(buffer);
- if (readableStreamHasDefaultReader(stream)) {
- if (readableStreamGetNumReadRequests(stream) === 0) {
- readableByteStreamControllerEnqueueChunkToQueue(
- controller,
- transferredBuffer,
- byteOffset,
- byteLength,
- );
- } else {
- assert(controller[sym.queue].length === 0);
- const transferredView = new Uint8Array(
- transferredBuffer,
- byteOffset,
- byteLength,
- );
- readableStreamFulfillReadRequest(stream, transferredView, false);
- }
- // 3.13.9.8 Otherwise, if ! ReadableStreamHasBYOBReader(stream) is true
- } else {
- assert(!isReadableStreamLocked(stream));
- readableByteStreamControllerEnqueueChunkToQueue(
- controller,
- transferredBuffer,
- byteOffset,
- byteLength,
- );
- }
- readableByteStreamControllerCallPullIfNeeded(controller);
- }
-
- function readableByteStreamControllerEnqueueChunkToQueue(
- controller,
- buffer,
- byteOffset,
- byteLength,
- ) {
- controller[sym.queue].push({
- value: buffer,
- offset: byteOffset,
- size: byteLength,
- });
- controller[sym.queueTotalSize] += byteLength;
- }
-
- function readableByteStreamControllerError(
- controller,
- e,
- ) {
- const stream = controller[sym.controlledReadableByteStream];
- if (stream[sym.state] !== "readable") {
- return;
- }
- // 3.13.11.3 Perform ! ReadableByteStreamControllerClearPendingPullIntos(controller).
- resetQueue(controller);
- readableByteStreamControllerClearAlgorithms(controller);
- readableStreamError(stream, e);
- }
-
- function readableByteStreamControllerGetDesiredSize(
- controller,
- ) {
- const stream = controller[sym.controlledReadableByteStream];
- const state = stream[sym.state];
- if (state === "errored") {
- return null;
- }
- if (state === "closed") {
- return 0;
- }
- return controller[sym.strategyHWM] - controller[sym.queueTotalSize];
- }
-
- function readableByteStreamControllerHandleQueueDrain(
- controller,
- ) {
- assert(
- controller[sym.controlledReadableByteStream][sym.state] === "readable",
- );
- if (
- controller[sym.queueTotalSize] === 0 && controller[sym.closeRequested]
- ) {
- readableByteStreamControllerClearAlgorithms(controller);
- readableStreamClose(controller[sym.controlledReadableByteStream]);
- } else {
- readableByteStreamControllerCallPullIfNeeded(controller);
- }
- }
-
- function readableStreamAddReadRequest(
- stream,
- ) {
- assert(isReadableStreamDefaultReader(stream[sym.reader]));
- assert(stream[sym.state] === "readable");
- const promise = getDeferred();
- stream[sym.reader][sym.readRequests].push(promise);
- return promise.promise;
- }
-
- function readableStreamCancel(
- stream,
- reason,
- ) {
- stream[sym.disturbed] = true;
- if (stream[sym.state] === "closed") {
- return Promise.resolve();
- }
- if (stream[sym.state] === "errored") {
- return Promise.reject(stream[sym.storedError]);
- }
- readableStreamClose(stream);
- return stream[sym.readableStreamController][sym.cancelSteps](reason).then(
- () => undefined,
- );
- }
-
- function readableStreamClose(stream) {
- assert(stream[sym.state] === "readable");
- stream[sym.state] = "closed";
- const reader = stream[sym.reader];
- if (!reader) {
- return;
- }
- if (isReadableStreamDefaultReader(reader)) {
- for (const readRequest of reader[sym.readRequests]) {
- assert(readRequest.resolve);
- readRequest.resolve(
- readableStreamCreateReadResult(
- undefined,
- true,
- reader[sym.forAuthorCode],
- ),
- );
- }
- reader[sym.readRequests] = [];
- }
- const resolve = reader[sym.closedPromise].resolve;
- assert(resolve);
- resolve();
- }
-
- function readableStreamCreateReadResult(
- value,
- done,
- forAuthorCode,
- ) {
- const prototype = forAuthorCode ? Object.prototype : null;
- assert(typeof done === "boolean");
- const obj = Object.create(prototype);
- Object.defineProperties(obj, {
- value: { value, writable: true, enumerable: true, configurable: true },
- done: {
- value: done,
- writable: true,
- enumerable: true,
- configurable: true,
- },
- });
- return obj;
- }
-
- function readableStreamDefaultControllerCallPullIfNeeded(
- controller,
- ) {
- const shouldPull = readableStreamDefaultControllerShouldCallPull(
- controller,
- );
- if (!shouldPull) {
- return;
- }
- if (controller[sym.pulling]) {
- controller[sym.pullAgain] = true;
- return;
- }
- assert(controller[sym.pullAgain] === false);
- controller[sym.pulling] = true;
- const pullPromise = controller[sym.pullAlgorithm]();
- pullPromise.then(
- () => {
- controller[sym.pulling] = false;
- if (controller[sym.pullAgain]) {
- controller[sym.pullAgain] = false;
- readableStreamDefaultControllerCallPullIfNeeded(controller);
- }
- },
- (e) => {
- readableStreamDefaultControllerError(controller, e);
- },
- );
- }
-
- function readableStreamDefaultControllerCanCloseOrEnqueue(
- controller,
- ) {
- const state = controller[sym.controlledReadableStream][sym.state];
- return !controller[sym.closeRequested] && state === "readable";
- }
-
- function readableStreamDefaultControllerClearAlgorithms(
- controller,
- ) {
- controller[sym.pullAlgorithm] = undefined;
- controller[sym.cancelAlgorithm] = undefined;
- controller[sym.strategySizeAlgorithm] = undefined;
- }
-
- function readableStreamDefaultControllerClose(
- controller,
- ) {
- if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller)) {
- return;
- }
- const stream = controller[sym.controlledReadableStream];
- controller[sym.closeRequested] = true;
- if (controller[sym.queue].length === 0) {
- readableStreamDefaultControllerClearAlgorithms(controller);
- readableStreamClose(stream);
- }
- }
-
- function readableStreamDefaultControllerEnqueue(
- controller,
- chunk,
- ) {
- if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller)) {
- return;
- }
- const stream = controller[sym.controlledReadableStream];
- if (
- isReadableStreamLocked(stream) &&
- readableStreamGetNumReadRequests(stream) > 0
- ) {
- readableStreamFulfillReadRequest(stream, chunk, false);
- } else {
- try {
- const chunkSize = controller[sym.strategySizeAlgorithm](chunk);
- enqueueValueWithSize(controller, chunk, chunkSize);
- } catch (err) {
- readableStreamDefaultControllerError(controller, err);
- throw err;
- }
- }
- readableStreamDefaultControllerCallPullIfNeeded(controller);
- }
-
- function readableStreamDefaultControllerGetDesiredSize(
- controller,
- ) {
- const stream = controller[sym.controlledReadableStream];
- const state = stream[sym.state];
- if (state === "errored") {
- return null;
- }
- if (state === "closed") {
- return 0;
- }
- return controller[sym.strategyHWM] - controller[sym.queueTotalSize];
- }
-
- function readableStreamDefaultControllerError(
- controller,
- e,
- ) {
- const stream = controller[sym.controlledReadableStream];
- if (stream[sym.state] !== "readable") {
- return;
- }
- resetQueue(controller);
- readableStreamDefaultControllerClearAlgorithms(controller);
- readableStreamError(stream, e);
- }
-
- function readableStreamDefaultControllerHasBackpressure(
- controller,
- ) {
- return readableStreamDefaultControllerShouldCallPull(controller);
- }
-
- function readableStreamDefaultControllerShouldCallPull(
- controller,
- ) {
- const stream = controller[sym.controlledReadableStream];
- if (
- !readableStreamDefaultControllerCanCloseOrEnqueue(controller) ||
- controller[sym.started] === false
- ) {
- return false;
- }
- if (
- isReadableStreamLocked(stream) &&
- readableStreamGetNumReadRequests(stream) > 0
- ) {
- return true;
- }
- const desiredSize = readableStreamDefaultControllerGetDesiredSize(
- controller,
- );
- assert(desiredSize !== null);
- return desiredSize > 0;
- }
-
- function readableStreamDefaultReaderRead(
- reader,
- ) {
- const stream = reader[sym.ownerReadableStream];
- assert(stream);
- stream[sym.disturbed] = true;
- if (stream[sym.state] === "closed") {
- return Promise.resolve(
- readableStreamCreateReadResult(
- undefined,
- true,
- reader[sym.forAuthorCode],
- ),
- );
- }
- if (stream[sym.state] === "errored") {
- return Promise.reject(stream[sym.storedError]);
- }
- assert(stream[sym.state] === "readable");
- return (stream[
- sym.readableStreamController
- ])[sym.pullSteps]();
- }
-
- function readableStreamError(stream, e) {
- assert(isReadableStream(stream));
- assert(stream[sym.state] === "readable");
- stream[sym.state] = "errored";
- stream[sym.storedError] = e;
- const reader = stream[sym.reader];
- if (reader === undefined) {
- return;
- }
- if (isReadableStreamDefaultReader(reader)) {
- for (const readRequest of reader[sym.readRequests]) {
- assert(readRequest.reject);
- readRequest.reject(e);
- readRequest.reject = undefined;
- readRequest.resolve = undefined;
- }
- reader[sym.readRequests] = [];
- }
- // 3.5.6.8 Otherwise, support BYOB Reader
- reader[sym.closedPromise].reject(e);
- reader[sym.closedPromise].reject = undefined;
- reader[sym.closedPromise].resolve = undefined;
- setPromiseIsHandledToTrue(reader[sym.closedPromise].promise);
- }
-
- function readableStreamFulfillReadRequest(
- stream,
- chunk,
- done,
- ) {
- const reader = stream[sym.reader];
- const readRequest = reader[sym.readRequests].shift();
- assert(readRequest.resolve);
- readRequest.resolve(
- readableStreamCreateReadResult(chunk, done, reader[sym.forAuthorCode]),
- );
- }
-
- function readableStreamGetNumReadRequests(
- stream,
- ) {
- return stream[sym.reader]?.[sym.readRequests].length ?? 0;
- }
-
- function readableStreamHasDefaultReader(
- stream,
- ) {
- const reader = stream[sym.reader];
- return !(reader === undefined || !isReadableStreamDefaultReader(reader));
- }
-
- function readableStreamPipeTo(
- source,
- dest,
- preventClose,
- preventAbort,
- preventCancel,
- signal,
- ) {
- assert(isReadableStream(source));
- assert(isWritableStream(dest));
- assert(
- typeof preventClose === "boolean" &&
- typeof preventAbort === "boolean" &&
- typeof preventCancel === "boolean",
- );
- assert(signal === undefined || signal instanceof AbortSignal);
- assert(!isReadableStreamLocked(source));
- assert(!isWritableStreamLocked(dest));
- const reader = acquireReadableStreamDefaultReader(source);
- const writer = acquireWritableStreamDefaultWriter(dest);
- source[sym.disturbed] = true;
- let shuttingDown = false;
- const promise = getDeferred();
- let abortAlgorithm;
- if (signal) {
- abortAlgorithm = () => {
- const error = new DOMException("Abort signal received.", "AbortSignal");
- const actions = [];
- if (!preventAbort) {
- actions.push(() => {
- if (dest[sym.state] === "writable") {
- return writableStreamAbort(dest, error);
- } else {
- return Promise.resolve(undefined);
- }
- });
- }
- if (!preventCancel) {
- actions.push(() => {
- if (source[sym.state] === "readable") {
- return readableStreamCancel(source, error);
- } else {
- return Promise.resolve(undefined);
- }
- });
- }
- shutdownWithAction(
- () => Promise.all(actions.map((action) => action())),
- true,
- error,
- );
- };
- if (signal.aborted) {
- abortAlgorithm();
- return promise.promise;
- }
- signal.addEventListener("abort", abortAlgorithm);
- }
-
- let currentWrite = Promise.resolve();
-
- // At this point, the spec becomes non-specific and vague. Most of the rest
- // of this code is based on the reference implementation that is part of the
- // specification. This is why the functions are only scoped to this function
- // to ensure they don't leak into the spec compliant parts.
-
- function isOrBecomesClosed(
- stream,
- promise,
- action,
- ) {
- if (stream[sym.state] === "closed") {
- action();
- } else {
- setPromiseIsHandledToTrue(promise.then(action));
- }
- }
-
- function isOrBecomesErrored(
- stream,
- promise,
- action,
- ) {
- if (stream[sym.state] === "errored") {
- action(stream[sym.storedError]);
- } else {
- setPromiseIsHandledToTrue(promise.catch((error) => action(error)));
- }
- }
-
- function finalize(isError, error) {
- writableStreamDefaultWriterRelease(writer);
- readableStreamReaderGenericRelease(reader);
-
- if (signal) {
- signal.removeEventListener("abort", abortAlgorithm);
- }
- if (isError) {
- promise.reject(error);
- } else {
- promise.resolve();
- }
- }
-
- function waitForWritesToFinish() {
- const oldCurrentWrite = currentWrite;
- return currentWrite.then(() =>
- oldCurrentWrite !== currentWrite ? waitForWritesToFinish() : undefined
- );
- }
-
- function shutdownWithAction(
- action,
- originalIsError,
- originalError,
- ) {
- function doTheRest() {
- setPromiseIsHandledToTrue(
- action().then(
- () => finalize(originalIsError, originalError),
- (newError) => finalize(true, newError),
- ),
- );
- }
-
- if (shuttingDown) {
- return;
- }
- shuttingDown = true;
-
- if (
- dest[sym.state] === "writable" &&
- writableStreamCloseQueuedOrInFlight(dest) === false
- ) {
- setPromiseIsHandledToTrue(waitForWritesToFinish().then(doTheRest));
- } else {
- doTheRest();
- }
- }
-
- function shutdown(isError, error) {
- if (shuttingDown) {
- return;
- }
- shuttingDown = true;
-
- if (
- dest[sym.state] === "writable" &&
- !writableStreamCloseQueuedOrInFlight(dest)
- ) {
- setPromiseIsHandledToTrue(
- waitForWritesToFinish().then(() => finalize(isError, error)),
- );
- }
- finalize(isError, error);
- }
-
- function pipeStep() {
- if (shuttingDown) {
- return Promise.resolve(true);
- }
- return writer[sym.readyPromise].promise.then(() => {
- return readableStreamDefaultReaderRead(reader).then(
- ({ value, done }) => {
- if (done === true) {
- return true;
- }
- currentWrite = writableStreamDefaultWriterWrite(
- writer,
- value,
- ).then(undefined, () => {});
- return false;
- },
- );
- });
- }
-
- function pipeLoop() {
- return new Promise((resolveLoop, rejectLoop) => {
- function next(done) {
- if (done) {
- resolveLoop(undefined);
- } else {
- setPromiseIsHandledToTrue(pipeStep().then(next, rejectLoop));
- }
- }
- next(false);
- });
- }
-
- isOrBecomesErrored(
- source,
- reader[sym.closedPromise].promise,
- (storedError) => {
- if (!preventAbort) {
- shutdownWithAction(
- () => writableStreamAbort(dest, storedError),
- true,
- storedError,
- );
- } else {
- shutdown(true, storedError);
- }
- },
- );
-
- isOrBecomesErrored(
- dest,
- writer[sym.closedPromise].promise,
- (storedError) => {
- if (!preventCancel) {
- shutdownWithAction(
- () => readableStreamCancel(source, storedError),
- true,
- storedError,
- );
- } else {
- shutdown(true, storedError);
- }
- },
- );
-
- isOrBecomesClosed(source, reader[sym.closedPromise].promise, () => {
- if (!preventClose) {
- shutdownWithAction(() =>
- writableStreamDefaultWriterCloseWithErrorPropagation(writer)
- );
- }
- });
-
- if (
- writableStreamCloseQueuedOrInFlight(dest) ||
- dest[sym.state] === "closed"
- ) {
- const destClosed = new TypeError(
- "The destination writable stream closed before all data could be piped to it.",
- );
- if (!preventCancel) {
- shutdownWithAction(
- () => readableStreamCancel(source, destClosed),
- true,
- destClosed,
- );
- } else {
- shutdown(true, destClosed);
- }
- }
-
- setPromiseIsHandledToTrue(pipeLoop());
- return promise.promise;
- }
-
- function readableStreamReaderGenericCancel(
- reader,
- reason,
- ) {
- const stream = reader[sym.ownerReadableStream];
- assert(stream);
- return readableStreamCancel(stream, reason);
- }
-
- function readableStreamReaderGenericInitialize(
- reader,
- stream,
- ) {
- reader[sym.forAuthorCode] = true;
- reader[sym.ownerReadableStream] = stream;
- stream[sym.reader] = reader;
- if (stream[sym.state] === "readable") {
- reader[sym.closedPromise] = getDeferred();
- } else if (stream[sym.state] === "closed") {
- reader[sym.closedPromise] = { promise: Promise.resolve() };
- } else {
- assert(stream[sym.state] === "errored");
- reader[sym.closedPromise] = {
- promise: Promise.reject(stream[sym.storedError]),
- };
- setPromiseIsHandledToTrue(reader[sym.closedPromise].promise);
- }
- }
-
- function readableStreamReaderGenericRelease(
- reader,
- ) {
- assert(reader[sym.ownerReadableStream]);
- assert(reader[sym.ownerReadableStream][sym.reader] === reader);
- const closedPromise = reader[sym.closedPromise];
- if (reader[sym.ownerReadableStream][sym.state] === "readable") {
- assert(closedPromise.reject);
- closedPromise.reject(new TypeError("ReadableStream state is readable."));
- } else {
- closedPromise.promise = Promise.reject(
- new TypeError("Reading is closed."),
- );
- delete closedPromise.reject;
- delete closedPromise.resolve;
- }
- setPromiseIsHandledToTrue(closedPromise.promise);
- reader[sym.ownerReadableStream][sym.reader] = undefined;
- reader[sym.ownerReadableStream] = undefined;
- }
-
- function readableStreamTee(
- stream,
- cloneForBranch2,
- ) {
- assert(isReadableStream(stream));
- assert(typeof cloneForBranch2 === "boolean");
- const reader = acquireReadableStreamDefaultReader(stream);
- let reading = false;
- let canceled1 = false;
- let canceled2 = false;
- let reason1 = undefined;
- let reason2 = undefined;
- /* eslint-disable prefer-const */
- let branch1;
- let branch2;
- /* eslint-enable prefer-const */
- const cancelPromise = getDeferred();
- const pullAlgorithm = () => {
- if (reading) {
- return Promise.resolve();
- }
- reading = true;
- const readPromise = readableStreamDefaultReaderRead(reader).then(
- (result) => {
- reading = false;
- assert(typeof result === "object");
- const { done } = result;
- assert(typeof done === "boolean");
- if (done) {
- if (!canceled1) {
- readableStreamDefaultControllerClose(
- branch1[
- sym.readableStreamController
- ],
- );
- }
- if (!canceled2) {
- readableStreamDefaultControllerClose(
- branch2[
- sym.readableStreamController
- ],
- );
- }
- return;
- }
- const { value } = result;
- const value1 = value;
- let value2 = value;
- if (!canceled2 && cloneForBranch2) {
- value2 = cloneValue(value2);
- }
- if (!canceled1) {
- readableStreamDefaultControllerEnqueue(
- branch1[
- sym.readableStreamController
- ],
- value1,
- );
- }
- if (!canceled2) {
- readableStreamDefaultControllerEnqueue(
- branch2[
- sym.readableStreamController
- ],
- value2,
- );
- }
- },
- );
- setPromiseIsHandledToTrue(readPromise);
- return Promise.resolve();
- };
- const cancel1Algorithm = (reason) => {
- canceled1 = true;
- reason1 = reason;
- if (canceled2) {
- const compositeReason = [reason1, reason2];
- const cancelResult = readableStreamCancel(stream, compositeReason);
- cancelPromise.resolve(cancelResult);
- }
- return cancelPromise.promise;
- };
- const cancel2Algorithm = (reason) => {
- canceled2 = true;
- reason2 = reason;
- if (canceled1) {
- const compositeReason = [reason1, reason2];
- const cancelResult = readableStreamCancel(stream, compositeReason);
- cancelPromise.resolve(cancelResult);
- }
- return cancelPromise.promise;
- };
- const startAlgorithm = () => undefined;
- branch1 = createReadableStream(
- startAlgorithm,
- pullAlgorithm,
- cancel1Algorithm,
- );
- branch2 = createReadableStream(
- startAlgorithm,
- pullAlgorithm,
- cancel2Algorithm,
- );
- setPromiseIsHandledToTrue(
- reader[sym.closedPromise].promise.catch((r) => {
- readableStreamDefaultControllerError(
- branch1[
- sym.readableStreamController
- ],
- r,
- );
- readableStreamDefaultControllerError(
- branch2[
- sym.readableStreamController
- ],
- r,
- );
- }),
- );
- return [branch1, branch2];
- }
-
- function resetQueue(container) {
- assert(sym.queue in container && sym.queueTotalSize in container);
- container[sym.queue] = [];
- container[sym.queueTotalSize] = 0;
- }
-
- /** An internal function which mimics the behavior of setting the promise to
- * handled in JavaScript. In this situation, an assertion failure, which
- * shouldn't happen will get thrown, instead of swallowed. */
- function setPromiseIsHandledToTrue(promise) {
- promise.then(undefined, (e) => {
- if (e && e instanceof AssertionError) {
- queueMicrotask(() => {
- throw e;
- });
- }
- });
- }
-
- function setUpReadableByteStreamController(
- stream,
- controller,
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark,
- autoAllocateChunkSize,
- ) {
- assert(stream[sym.readableStreamController] === undefined);
- if (autoAllocateChunkSize !== undefined) {
- assert(Number.isInteger(autoAllocateChunkSize));
- assert(autoAllocateChunkSize >= 0);
- }
- controller[sym.controlledReadableByteStream] = stream;
- controller[sym.pulling] = controller[sym.pullAgain] = false;
- controller[sym.byobRequest] = undefined;
- controller[sym.queue] = [];
- controller[sym.queueTotalSize] = 0;
- controller[sym.closeRequested] = controller[sym.started] = false;
- controller[sym.strategyHWM] = validateAndNormalizeHighWaterMark(
- highWaterMark,
- );
- controller[sym.pullAlgorithm] = pullAlgorithm;
- controller[sym.cancelAlgorithm] = cancelAlgorithm;
- controller[sym.autoAllocateChunkSize] = autoAllocateChunkSize;
- // 3.13.26.12 Set controller.[[pendingPullIntos]] to a new empty List.
- stream[sym.readableStreamController] = controller;
- const startResult = startAlgorithm();
- const startPromise = Promise.resolve(startResult);
- setPromiseIsHandledToTrue(
- startPromise.then(
- () => {
- controller[sym.started] = true;
- assert(!controller[sym.pulling]);
- assert(!controller[sym.pullAgain]);
- readableByteStreamControllerCallPullIfNeeded(controller);
- },
- (r) => {
- readableByteStreamControllerError(controller, r);
- },
- ),
- );
- }
-
- function setUpReadableByteStreamControllerFromUnderlyingSource(
- stream,
- underlyingByteSource,
- highWaterMark,
- ) {
- assert(underlyingByteSource);
- const controller = Object.create(
- ReadableByteStreamController.prototype,
- );
- const startAlgorithm = () => {
- return invokeOrNoop(underlyingByteSource, "start", controller);
- };
- const pullAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingByteSource,
- "pull",
- 0,
- controller,
- );
- setFunctionName(pullAlgorithm, "[[pullAlgorithm]]");
- const cancelAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingByteSource,
- "cancel",
- 1,
- );
- setFunctionName(cancelAlgorithm, "[[cancelAlgorithm]]");
- // 3.13.27.6 Let autoAllocateChunkSize be ? GetV(underlyingByteSource, "autoAllocateChunkSize").
- const autoAllocateChunkSize = undefined;
- setUpReadableByteStreamController(
- stream,
- controller,
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark,
- autoAllocateChunkSize,
- );
- }
-
- function setUpReadableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- ) {
- assert(stream[sym.readableStreamController] === undefined);
- controller[sym.controlledReadableStream] = stream;
- controller[sym.queue] = [];
- controller[sym.queueTotalSize] = 0;
- controller[sym.started] = controller[sym.closeRequested] = controller[
- sym.pullAgain
- ] = controller[sym.pulling] = false;
- controller[sym.strategySizeAlgorithm] = sizeAlgorithm;
- controller[sym.strategyHWM] = highWaterMark;
- controller[sym.pullAlgorithm] = pullAlgorithm;
- controller[sym.cancelAlgorithm] = cancelAlgorithm;
- stream[sym.readableStreamController] = controller;
- const startResult = startAlgorithm();
- const startPromise = Promise.resolve(startResult);
- setPromiseIsHandledToTrue(
- startPromise.then(
- () => {
- controller[sym.started] = true;
- assert(controller[sym.pulling] === false);
- assert(controller[sym.pullAgain] === false);
- readableStreamDefaultControllerCallPullIfNeeded(controller);
- },
- (r) => {
- readableStreamDefaultControllerError(controller, r);
- },
- ),
- );
- }
-
- function setUpReadableStreamDefaultControllerFromUnderlyingSource(
- stream,
- underlyingSource,
- highWaterMark,
- sizeAlgorithm,
- ) {
- assert(underlyingSource);
- const controller = Object.create(
- ReadableStreamDefaultController.prototype,
- );
- const startAlgorithm = () =>
- invokeOrNoop(underlyingSource, "start", controller);
- const pullAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingSource,
- "pull",
- 0,
- controller,
- );
- setFunctionName(pullAlgorithm, "[[pullAlgorithm]]");
- const cancelAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingSource,
- "cancel",
- 1,
- );
- setFunctionName(cancelAlgorithm, "[[cancelAlgorithm]]");
- setUpReadableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- pullAlgorithm,
- cancelAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- );
- }
-
- function setUpTransformStreamDefaultController(
- stream,
- controller,
- transformAlgorithm,
- flushAlgorithm,
- ) {
- assert(isTransformStream(stream));
- assert(stream[sym.transformStreamController] === undefined);
- controller[sym.controlledTransformStream] = stream;
- stream[sym.transformStreamController] = controller;
- controller[sym.transformAlgorithm] = transformAlgorithm;
- controller[sym.flushAlgorithm] = flushAlgorithm;
- }
-
- function setUpTransformStreamDefaultControllerFromTransformer(
- stream,
- transformer,
- ) {
- assert(transformer);
- const controller = Object.create(
- TransformStreamDefaultController.prototype,
- );
- let transformAlgorithm = (chunk) => {
- try {
- transformStreamDefaultControllerEnqueue(
- controller,
- // it defaults to no tranformation, so I is assumed to be O
- chunk,
- );
- } catch (e) {
- return Promise.reject(e);
- }
- return Promise.resolve();
- };
- const transformMethod = transformer.transform;
- if (transformMethod) {
- if (typeof transformMethod !== "function") {
- throw new TypeError("tranformer.transform must be callable.");
- }
- transformAlgorithm = async (chunk) =>
- call(transformMethod, transformer, [chunk, controller]);
- }
- const flushAlgorithm = createAlgorithmFromUnderlyingMethod(
- transformer,
- "flush",
- 0,
- controller,
- );
- setUpTransformStreamDefaultController(
- stream,
- controller,
- transformAlgorithm,
- flushAlgorithm,
- );
- }
-
- function setUpWritableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- writeAlgorithm,
- closeAlgorithm,
- abortAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- ) {
- assert(isWritableStream(stream));
- assert(stream[sym.writableStreamController] === undefined);
- controller[sym.controlledWritableStream] = stream;
- stream[sym.writableStreamController] = controller;
- controller[sym.queue] = [];
- controller[sym.queueTotalSize] = 0;
- controller[sym.started] = false;
- controller[sym.strategySizeAlgorithm] = sizeAlgorithm;
- controller[sym.strategyHWM] = highWaterMark;
- controller[sym.writeAlgorithm] = writeAlgorithm;
- controller[sym.closeAlgorithm] = closeAlgorithm;
- controller[sym.abortAlgorithm] = abortAlgorithm;
- const backpressure = writableStreamDefaultControllerGetBackpressure(
- controller,
- );
- writableStreamUpdateBackpressure(stream, backpressure);
- const startResult = startAlgorithm();
- const startPromise = Promise.resolve(startResult);
- setPromiseIsHandledToTrue(
- startPromise.then(
- () => {
- assert(
- stream[sym.state] === "writable" ||
- stream[sym.state] === "erroring",
- );
- controller[sym.started] = true;
- writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
- },
- (r) => {
- assert(
- stream[sym.state] === "writable" ||
- stream[sym.state] === "erroring",
- );
- controller[sym.started] = true;
- writableStreamDealWithRejection(stream, r);
- },
- ),
- );
- }
-
- function setUpWritableStreamDefaultControllerFromUnderlyingSink(
- stream,
- underlyingSink,
- highWaterMark,
- sizeAlgorithm,
- ) {
- assert(underlyingSink);
- const controller = Object.create(
- WritableStreamDefaultController.prototype,
- );
- const startAlgorithm = () => {
- return invokeOrNoop(underlyingSink, "start", controller);
- };
- const writeAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingSink,
- "write",
- 1,
- controller,
- );
- setFunctionName(writeAlgorithm, "[[writeAlgorithm]]");
- const closeAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingSink,
- "close",
- 0,
- );
- setFunctionName(closeAlgorithm, "[[closeAlgorithm]]");
- const abortAlgorithm = createAlgorithmFromUnderlyingMethod(
- underlyingSink,
- "abort",
- 1,
- );
- setFunctionName(abortAlgorithm, "[[abortAlgorithm]]");
- setUpWritableStreamDefaultController(
- stream,
- controller,
- startAlgorithm,
- writeAlgorithm,
- closeAlgorithm,
- abortAlgorithm,
- highWaterMark,
- sizeAlgorithm,
- );
- }
-
- function transformStreamDefaultControllerClearAlgorithms(
- controller,
- ) {
- controller[sym.transformAlgorithm] = undefined;
- controller[sym.flushAlgorithm] = undefined;
- }
-
- function transformStreamDefaultControllerEnqueue(
- controller,
- chunk,
- ) {
- const stream = controller[sym.controlledTransformStream];
- const readableController = stream[sym.readable][
- sym.readableStreamController
- ];
- if (!readableStreamDefaultControllerCanCloseOrEnqueue(readableController)) {
- throw new TypeError(
- "TransformStream's readable controller cannot be closed or enqueued.",
- );
- }
- try {
- readableStreamDefaultControllerEnqueue(readableController, chunk);
- } catch (e) {
- transformStreamErrorWritableAndUnblockWrite(stream, e);
- throw stream[sym.readable][sym.storedError];
- }
- const backpressure = readableStreamDefaultControllerHasBackpressure(
- readableController,
- );
- if (backpressure) {
- transformStreamSetBackpressure(stream, true);
- }
- }
-
- function transformStreamDefaultControllerError(
- controller,
- e,
- ) {
- transformStreamError(controller[sym.controlledTransformStream], e);
- }
-
- function transformStreamDefaultControllerPerformTransform(
- controller,
- chunk,
- ) {
- const transformPromise = controller[sym.transformAlgorithm](chunk);
- return transformPromise.then(undefined, (r) => {
- transformStreamError(controller[sym.controlledTransformStream], r);
- throw r;
- });
- }
-
- function transformStreamDefaultSinkAbortAlgorithm(
- stream,
- reason,
- ) {
- transformStreamError(stream, reason);
- return Promise.resolve(undefined);
- }
-
- function transformStreamDefaultSinkCloseAlgorithm(
- stream,
- ) {
- const readable = stream[sym.readable];
- const controller = stream[sym.transformStreamController];
- const flushPromise = controller[sym.flushAlgorithm]();
- transformStreamDefaultControllerClearAlgorithms(controller);
- return flushPromise.then(
- () => {
- if (readable[sym.state] === "errored") {
- throw readable[sym.storedError];
- }
- const readableController = readable[
- sym.readableStreamController
- ];
- if (
- readableStreamDefaultControllerCanCloseOrEnqueue(readableController)
- ) {
- readableStreamDefaultControllerClose(readableController);
- }
- },
- (r) => {
- transformStreamError(stream, r);
- throw readable[sym.storedError];
- },
- );
- }
-
- function transformStreamDefaultSinkWriteAlgorithm(
- stream,
- chunk,
- ) {
- assert(stream[sym.writable][sym.state] === "writable");
- const controller = stream[sym.transformStreamController];
- if (stream[sym.backpressure]) {
- const backpressureChangePromise = stream[sym.backpressureChangePromise];
- assert(backpressureChangePromise);
- return backpressureChangePromise.promise.then(() => {
- const writable = stream[sym.writable];
- const state = writable[sym.state];
- if (state === "erroring") {
- throw writable[sym.storedError];
- }
- assert(state === "writable");
- return transformStreamDefaultControllerPerformTransform(
- controller,
- chunk,
- );
- });
- }
- return transformStreamDefaultControllerPerformTransform(controller, chunk);
- }
-
- function transformStreamDefaultSourcePullAlgorithm(
- stream,
- ) {
- assert(stream[sym.backpressure] === true);
- assert(stream[sym.backpressureChangePromise] !== undefined);
- transformStreamSetBackpressure(stream, false);
- return stream[sym.backpressureChangePromise].promise;
- }
-
- function transformStreamError(
- stream,
- e,
- ) {
- readableStreamDefaultControllerError(
- stream[sym.readable][
- sym.readableStreamController
- ],
- e,
- );
- transformStreamErrorWritableAndUnblockWrite(stream, e);
- }
-
- function transformStreamDefaultControllerTerminate(
- controller,
- ) {
- const stream = controller[sym.controlledTransformStream];
- const readableController = stream[sym.readable][
- sym.readableStreamController
- ];
- readableStreamDefaultControllerClose(readableController);
- const error = new TypeError("TransformStream is closed.");
- transformStreamErrorWritableAndUnblockWrite(stream, error);
- }
-
- function transformStreamErrorWritableAndUnblockWrite(
- stream,
- e,
- ) {
- transformStreamDefaultControllerClearAlgorithms(
- stream[sym.transformStreamController],
- );
- writableStreamDefaultControllerErrorIfNeeded(
- stream[sym.writable][sym.writableStreamController],
- e,
- );
- if (stream[sym.backpressure]) {
- transformStreamSetBackpressure(stream, false);
- }
- }
-
- function transformStreamSetBackpressure(
- stream,
- backpressure,
- ) {
- assert(stream[sym.backpressure] !== backpressure);
- if (stream[sym.backpressureChangePromise] !== undefined) {
- stream[sym.backpressureChangePromise].resolve(undefined);
- }
- stream[sym.backpressureChangePromise] = getDeferred();
- stream[sym.backpressure] = backpressure;
- }
-
- function transferArrayBuffer(buffer) {
- assert(!isDetachedBuffer(buffer));
- const transferredIshVersion = buffer.slice(0);
-
- Object.defineProperty(buffer, "byteLength", {
- get() {
- return 0;
- },
- });
- buffer[sym.isFakeDetached] = true;
-
- return transferredIshVersion;
- }
-
- function validateAndNormalizeHighWaterMark(
- highWaterMark,
- ) {
- highWaterMark = Number(highWaterMark);
- if (Number.isNaN(highWaterMark) || highWaterMark < 0) {
- throw new RangeError(
- `highWaterMark must be a positive number or Infinity. Received: ${highWaterMark}.`,
- );
- }
- return highWaterMark;
- }
-
- function writableStreamAbort(
- stream,
- reason,
- ) {
- const state = stream[sym.state];
- if (state === "closed" || state === "errored") {
- return Promise.resolve(undefined);
- }
- if (stream[sym.pendingAbortRequest]) {
- return stream[sym.pendingAbortRequest].promise.promise;
- }
- assert(state === "writable" || state === "erroring");
- let wasAlreadyErroring = false;
- if (state === "erroring") {
- wasAlreadyErroring = true;
- reason = undefined;
- }
- const promise = getDeferred();
- stream[sym.pendingAbortRequest] = { promise, reason, wasAlreadyErroring };
-
- if (wasAlreadyErroring === false) {
- writableStreamStartErroring(stream, reason);
- }
- return promise.promise;
- }
-
- function writableStreamAddWriteRequest(
- stream,
- ) {
- assert(isWritableStream(stream));
- assert(stream[sym.state] === "writable");
- const promise = getDeferred();
- stream[sym.writeRequests].push(promise);
- return promise.promise;
- }
-
- function writableStreamClose(
- stream,
- ) {
- const state = stream[sym.state];
- if (state === "closed" || state === "errored") {
- return Promise.reject(
- new TypeError(
- "Cannot close an already closed or errored WritableStream.",
- ),
- );
- }
- assert(!writableStreamCloseQueuedOrInFlight(stream));
- const promise = getDeferred();
- stream[sym.closeRequest] = promise;
- const writer = stream[sym.writer];
- if (writer && stream[sym.backpressure] && state === "writable") {
- writer[sym.readyPromise].resolve();
- writer[sym.readyPromise].resolve = undefined;
- writer[sym.readyPromise].reject = undefined;
- }
- writableStreamDefaultControllerClose(stream[sym.writableStreamController]);
- return promise.promise;
- }
-
- function writableStreamCloseQueuedOrInFlight(
- stream,
- ) {
- return !(
- stream[sym.closeRequest] === undefined &&
- stream[sym.inFlightCloseRequest] === undefined
- );
- }
-
- function writableStreamDealWithRejection(
- stream,
- error,
- ) {
- const state = stream[sym.state];
- if (state === "writable") {
- writableStreamStartErroring(stream, error);
- return;
- }
- assert(state === "erroring");
- writableStreamFinishErroring(stream);
- }
-
- function writableStreamDefaultControllerAdvanceQueueIfNeeded(
- controller,
- ) {
- const stream = controller[sym.controlledWritableStream];
- if (!controller[sym.started]) {
- return;
- }
- if (stream[sym.inFlightWriteRequest]) {
- return;
- }
- const state = stream[sym.state];
- assert(state !== "closed" && state !== "errored");
- if (state === "erroring") {
- writableStreamFinishErroring(stream);
- return;
- }
- if (!controller[sym.queue].length) {
- return;
- }
- const writeRecord = peekQueueValue(controller);
- if (writeRecord === "close") {
- writableStreamDefaultControllerProcessClose(controller);
- } else {
- writableStreamDefaultControllerProcessWrite(
- controller,
- writeRecord.chunk,
- );
- }
- }
-
- function writableStreamDefaultControllerClearAlgorithms(
- controller,
- ) {
- controller[sym.writeAlgorithm] = undefined;
- controller[sym.closeAlgorithm] = undefined;
- controller[sym.abortAlgorithm] = undefined;
- controller[sym.strategySizeAlgorithm] = undefined;
- }
-
- function writableStreamDefaultControllerClose(
- controller,
- ) {
- enqueueValueWithSize(controller, "close", 0);
- writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
- }
-
- function writableStreamDefaultControllerError(
- controller,
- error,
- ) {
- const stream = controller[sym.controlledWritableStream];
- assert(stream[sym.state] === "writable");
- writableStreamDefaultControllerClearAlgorithms(controller);
- writableStreamStartErroring(stream, error);
- }
-
- function writableStreamDefaultControllerErrorIfNeeded(
- controller,
- error,
- ) {
- if (controller[sym.controlledWritableStream][sym.state] === "writable") {
- writableStreamDefaultControllerError(controller, error);
- }
- }
-
- function writableStreamDefaultControllerGetBackpressure(
- controller,
- ) {
- const desiredSize = writableStreamDefaultControllerGetDesiredSize(
- controller,
- );
- return desiredSize <= 0;
- }
-
- function writableStreamDefaultControllerGetChunkSize(
- controller,
- chunk,
- ) {
- let returnValue;
- try {
- returnValue = controller[sym.strategySizeAlgorithm](chunk);
- } catch (e) {
- writableStreamDefaultControllerErrorIfNeeded(controller, e);
- return 1;
- }
- return returnValue;
- }
-
- function writableStreamDefaultControllerGetDesiredSize(
- controller,
- ) {
- return controller[sym.strategyHWM] - controller[sym.queueTotalSize];
- }
-
- function writableStreamDefaultControllerProcessClose(
- controller,
- ) {
- const stream = controller[sym.controlledWritableStream];
- writableStreamMarkCloseRequestInFlight(stream);
- dequeueValue(controller);
- assert(controller[sym.queue].length === 0);
- const sinkClosePromise = controller[sym.closeAlgorithm]();
- writableStreamDefaultControllerClearAlgorithms(controller);
- setPromiseIsHandledToTrue(
- sinkClosePromise.then(
- () => {
- writableStreamFinishInFlightClose(stream);
- },
- (reason) => {
- writableStreamFinishInFlightCloseWithError(stream, reason);
- },
- ),
- );
- }
-
- function writableStreamDefaultControllerProcessWrite(
- controller,
- chunk,
- ) {
- const stream = controller[sym.controlledWritableStream];
- writableStreamMarkFirstWriteRequestInFlight(stream);
- const sinkWritePromise = controller[sym.writeAlgorithm](chunk);
- setPromiseIsHandledToTrue(
- sinkWritePromise.then(
- () => {
- writableStreamFinishInFlightWrite(stream);
- const state = stream[sym.state];
- assert(state === "writable" || state === "erroring");
- dequeueValue(controller);
- if (
- !writableStreamCloseQueuedOrInFlight(stream) &&
- state === "writable"
- ) {
- const backpressure = writableStreamDefaultControllerGetBackpressure(
- controller,
- );
- writableStreamUpdateBackpressure(stream, backpressure);
- }
- writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
- },
- (reason) => {
- if (stream[sym.state] === "writable") {
- writableStreamDefaultControllerClearAlgorithms(controller);
- }
- writableStreamFinishInFlightWriteWithError(stream, reason);
- },
- ),
- );
- }
-
- function writableStreamDefaultControllerWrite(
- controller,
- chunk,
- chunkSize,
- ) {
- const writeRecord = { chunk };
- try {
- enqueueValueWithSize(controller, writeRecord, chunkSize);
- } catch (e) {
- writableStreamDefaultControllerErrorIfNeeded(controller, e);
- return;
- }
- const stream = controller[sym.controlledWritableStream];
- if (
- !writableStreamCloseQueuedOrInFlight(stream) &&
- stream[sym.state] === "writable"
- ) {
- const backpressure = writableStreamDefaultControllerGetBackpressure(
- controller,
- );
- writableStreamUpdateBackpressure(stream, backpressure);
- }
- writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
- }
-
- function writableStreamDefaultWriterAbort(
- writer,
- reason,
- ) {
- const stream = writer[sym.ownerWritableStream];
- assert(stream);
- return writableStreamAbort(stream, reason);
- }
-
- function writableStreamDefaultWriterClose(
- writer,
- ) {
- const stream = writer[sym.ownerWritableStream];
- assert(stream);
- return writableStreamClose(stream);
- }
-
- function writableStreamDefaultWriterCloseWithErrorPropagation(
- writer,
- ) {
- const stream = writer[sym.ownerWritableStream];
- assert(stream);
- const state = stream[sym.state];
- if (writableStreamCloseQueuedOrInFlight(stream) || state === "closed") {
- return Promise.resolve();
- }
- if (state === "errored") {
- return Promise.reject(stream[sym.storedError]);
- }
- assert(state === "writable" || state === "erroring");
- return writableStreamDefaultWriterClose(writer);
- }
-
- function writableStreamDefaultWriterEnsureClosePromiseRejected(
- writer,
- error,
- ) {
- if (writer[sym.closedPromise].reject) {
- writer[sym.closedPromise].reject(error);
- } else {
- writer[sym.closedPromise] = {
- promise: Promise.reject(error),
- };
- }
- setPromiseIsHandledToTrue(writer[sym.closedPromise].promise);
- }
-
- function writableStreamDefaultWriterEnsureReadyPromiseRejected(
- writer,
- error,
- ) {
- if (writer[sym.readyPromise].reject) {
- writer[sym.readyPromise].reject(error);
- writer[sym.readyPromise].reject = undefined;
- writer[sym.readyPromise].resolve = undefined;
- } else {
- writer[sym.readyPromise] = {
- promise: Promise.reject(error),
- };
- }
- setPromiseIsHandledToTrue(writer[sym.readyPromise].promise);
- }
-
- function writableStreamDefaultWriterWrite(
- writer,
- chunk,
- ) {
- const stream = writer[sym.ownerWritableStream];
- assert(stream);
- const controller = stream[sym.writableStreamController];
- assert(controller);
- const chunkSize = writableStreamDefaultControllerGetChunkSize(
- controller,
- chunk,
- );
- if (stream !== writer[sym.ownerWritableStream]) {
- return Promise.reject("Writer has incorrect WritableStream.");
- }
- const state = stream[sym.state];
- if (state === "errored") {
- return Promise.reject(stream[sym.storedError]);
- }
- if (writableStreamCloseQueuedOrInFlight(stream) || state === "closed") {
- return Promise.reject(new TypeError("The stream is closed or closing."));
- }
- if (state === "erroring") {
- return Promise.reject(stream[sym.storedError]);
- }
- assert(state === "writable");
- const promise = writableStreamAddWriteRequest(stream);
- writableStreamDefaultControllerWrite(controller, chunk, chunkSize);
- return promise;
- }
-
- function writableStreamDefaultWriterGetDesiredSize(
- writer,
- ) {
- const stream = writer[sym.ownerWritableStream];
- const state = stream[sym.state];
- if (state === "errored" || state === "erroring") {
- return null;
- }
- if (state === "closed") {
- return 0;
- }
- return writableStreamDefaultControllerGetDesiredSize(
- stream[sym.writableStreamController],
- );
- }
-
- function writableStreamDefaultWriterRelease(
- writer,
- ) {
- const stream = writer[sym.ownerWritableStream];
- assert(stream);
- assert(stream[sym.writer] === writer);
- const releasedError = new TypeError(
- "Writer was released and can no longer be used to monitor the stream's closedness.",
- );
- writableStreamDefaultWriterEnsureReadyPromiseRejected(
- writer,
- releasedError,
- );
- writableStreamDefaultWriterEnsureClosePromiseRejected(
- writer,
- releasedError,
- );
- stream[sym.writer] = undefined;
- writer[sym.ownerWritableStream] = undefined;
- }
-
- function writableStreamFinishErroring(stream) {
- assert(stream[sym.state] === "erroring");
- assert(!writableStreamHasOperationMarkedInFlight(stream));
- stream[sym.state] = "errored";
- stream[sym.writableStreamController][sym.errorSteps]();
- const storedError = stream[sym.storedError];
- for (const writeRequest of stream[sym.writeRequests]) {
- assert(writeRequest.reject);
- writeRequest.reject(storedError);
- }
- stream[sym.writeRequests] = [];
- if (!stream[sym.pendingAbortRequest]) {
- writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
- return;
- }
- const abortRequest = stream[sym.pendingAbortRequest];
- assert(abortRequest);
- stream[sym.pendingAbortRequest] = undefined;
- if (abortRequest.wasAlreadyErroring) {
- assert(abortRequest.promise.reject);
- abortRequest.promise.reject(storedError);
- writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
- return;
- }
- const promise = stream[sym.writableStreamController][sym.abortSteps](
- abortRequest.reason,
- );
- setPromiseIsHandledToTrue(
- promise.then(
- () => {
- assert(abortRequest.promise.resolve);
- abortRequest.promise.resolve();
- writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
- },
- (reason) => {
- assert(abortRequest.promise.reject);
- abortRequest.promise.reject(reason);
- writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
- },
- ),
- );
- }
-
- function writableStreamFinishInFlightClose(
- stream,
- ) {
- assert(stream[sym.inFlightCloseRequest]);
- stream[sym.inFlightCloseRequest]?.resolve();
- stream[sym.inFlightCloseRequest] = undefined;
- const state = stream[sym.state];
- assert(state === "writable" || state === "erroring");
- if (state === "erroring") {
- stream[sym.storedError] = undefined;
- if (stream[sym.pendingAbortRequest]) {
- stream[sym.pendingAbortRequest].promise.resolve();
- stream[sym.pendingAbortRequest] = undefined;
- }
- }
- stream[sym.state] = "closed";
- const writer = stream[sym.writer];
- if (writer) {
- writer[sym.closedPromise].resolve();
- }
- assert(stream[sym.pendingAbortRequest] === undefined);
- assert(stream[sym.storedError] === undefined);
- }
-
- function writableStreamFinishInFlightCloseWithError(
- stream,
- error,
- ) {
- assert(stream[sym.inFlightCloseRequest]);
- stream[sym.inFlightCloseRequest]?.reject(error);
- stream[sym.inFlightCloseRequest] = undefined;
- assert(
- stream[sym.state] === "writable" || stream[sym.state] === "erroring",
- );
- if (stream[sym.pendingAbortRequest]) {
- stream[sym.pendingAbortRequest]?.promise.reject(error);
- stream[sym.pendingAbortRequest] = undefined;
- }
- writableStreamDealWithRejection(stream, error);
- }
-
- function writableStreamFinishInFlightWrite(
- stream,
- ) {
- assert(stream[sym.inFlightWriteRequest]);
- stream[sym.inFlightWriteRequest].resolve();
- stream[sym.inFlightWriteRequest] = undefined;
- }
-
- function writableStreamFinishInFlightWriteWithError(
- stream,
- error,
- ) {
- assert(stream[sym.inFlightWriteRequest]);
- stream[sym.inFlightWriteRequest].reject(error);
- stream[sym.inFlightWriteRequest] = undefined;
- assert(
- stream[sym.state] === "writable" || stream[sym.state] === "erroring",
- );
- writableStreamDealWithRejection(stream, error);
- }
-
- function writableStreamHasOperationMarkedInFlight(
- stream,
- ) {
- return !(
- stream[sym.inFlightWriteRequest] === undefined &&
- stream[sym.inFlightCloseRequest] === undefined
- );
- }
-
- function writableStreamMarkCloseRequestInFlight(
- stream,
- ) {
- assert(stream[sym.inFlightCloseRequest] === undefined);
- assert(stream[sym.closeRequest] !== undefined);
- stream[sym.inFlightCloseRequest] = stream[sym.closeRequest];
- stream[sym.closeRequest] = undefined;
- }
-
- function writableStreamMarkFirstWriteRequestInFlight(
- stream,
- ) {
- assert(stream[sym.inFlightWriteRequest] === undefined);
- assert(stream[sym.writeRequests].length);
- const writeRequest = stream[sym.writeRequests].shift();
- stream[sym.inFlightWriteRequest] = writeRequest;
- }
-
- function writableStreamRejectCloseAndClosedPromiseIfNeeded(
- stream,
- ) {
- assert(stream[sym.state] === "errored");
- if (stream[sym.closeRequest]) {
- assert(stream[sym.inFlightCloseRequest] === undefined);
- stream[sym.closeRequest].reject(stream[sym.storedError]);
- stream[sym.closeRequest] = undefined;
- }
- const writer = stream[sym.writer];
- if (writer) {
- writer[sym.closedPromise].reject(stream[sym.storedError]);
- setPromiseIsHandledToTrue(writer[sym.closedPromise].promise);
- }
- }
-
- function writableStreamStartErroring(
- stream,
- reason,
- ) {
- assert(stream[sym.storedError] === undefined);
- assert(stream[sym.state] === "writable");
- const controller = stream[sym.writableStreamController];
- assert(controller);
- stream[sym.state] = "erroring";
- stream[sym.storedError] = reason;
- const writer = stream[sym.writer];
- if (writer) {
- writableStreamDefaultWriterEnsureReadyPromiseRejected(writer, reason);
- }
- if (
- !writableStreamHasOperationMarkedInFlight(stream) &&
- controller[sym.started]
- ) {
- writableStreamFinishErroring(stream);
- }
- }
-
- function writableStreamUpdateBackpressure(
- stream,
- backpressure,
- ) {
- assert(stream[sym.state] === "writable");
- assert(!writableStreamCloseQueuedOrInFlight(stream));
- const writer = stream[sym.writer];
- if (writer && backpressure !== stream[sym.backpressure]) {
- if (backpressure) {
- writer[sym.readyPromise] = getDeferred();
- } else {
- assert(backpressure === false);
- writer[sym.readyPromise].resolve();
- writer[sym.readyPromise].resolve = undefined;
- writer[sym.readyPromise].reject = undefined;
- }
- }
- stream[sym.backpressure] = backpressure;
- }
- /* eslint-enable */
-
- window.__bootstrap.streams = {
- ReadableStream,
- TransformStream,
- WritableStream,
- isReadableStreamDisturbed,
- };
-})(this);
generated by cgit v1.2.3 (git 2.25.1) at 2025-07-26 06:46:22 +0000