'use strict';
const {
ArrayBuffer,
ArrayBufferPrototypeGetByteLength,
ArrayBufferPrototypeGetDetached,
ArrayBufferPrototypeSlice,
ArrayBufferPrototypeTransfer,
ArrayPrototypePush,
AsyncIteratorPrototype,
DataView,
FunctionPrototypeBind,
FunctionPrototypeCall,
MathMin,
NumberIsInteger,
ObjectDefineProperties,
ObjectSetPrototypeOf,
Promise,
PromisePrototypeThen,
PromiseReject,
PromiseResolve,
PromiseWithResolvers,
SafePromiseAll,
Symbol,
SymbolAsyncIterator,
SymbolDispose,
SymbolIterator,
SymbolToStringTag,
TypedArrayPrototypeGetLength,
Uint8Array,
} = primordials;
const {
AbortError,
codes: {
ERR_ARG_NOT_ITERABLE,
ERR_ILLEGAL_CONSTRUCTOR,
ERR_INVALID_ARG_TYPE,
ERR_INVALID_ARG_VALUE,
ERR_INVALID_STATE,
ERR_INVALID_THIS,
ERR_OUT_OF_RANGE,
},
} = require('internal/errors');
const {
DOMException,
} = internalBinding('messaging');
const AsyncContextFrame = require('internal/async_context_frame');
const {
isArrayBufferView,
isDataView,
isSharedArrayBuffer,
} = require('internal/util/types');
const {
customInspectSymbol: kInspect,
kEmptyObject,
kEnumerableProperty,
SideEffectFreeRegExpPrototypeSymbolReplace,
} = require('internal/util');
const {
validateAbortSignal,
validateBuffer,
validateNumber,
validateObject,
kValidateObjectAllowObjects,
kValidateObjectAllowObjectsAndNull,
} = require('internal/validators');
const {
MessageChannel,
} = require('internal/worker/io');
const {
kDeserialize,
kTransfer,
kTransferList,
markTransferMode,
structuredClone,
} = require('internal/worker/js_transferable');
const {
queueMicrotask,
} = require('internal/process/task_queues');
const {
kIsDisturbed,
kIsErrored,
kIsReadable,
kIsClosedPromise,
kControllerErrorFunction,
} = require('internal/streams/utils');
const {
ArrayBufferViewGetBuffer,
ArrayBufferViewGetByteLength,
ArrayBufferViewGetByteOffset,
Queue,
canCopyArrayBuffer,
cloneAsUint8Array,
copyArrayBuffer,
createPromiseCallback1Param,
createRawCallback1Param,
customInspect,
defaultSizeAlgorithm,
dequeueValue,
enqueueValueWithSize,
extractHighWaterMark,
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
kDeferredStart,
kEmptyQueue,
kParkedAlgorithmResult,
kPendingPromise,
kResolvedPromise,
kState,
kType,
lazyTransfer,
materializeQueue,
nonOpCallback,
nonOpCancel,
rejectedHandledRecord,
resetQueue,
resolvedRecord,
setPromiseHandled,
thenAlgorithmResult,
} = require('internal/webstreams/util');
const {
WritableStreamDefaultWriter,
isWritableStream,
isWritableStreamLocked,
isWritableStreamDefaultController,
isWritableStreamDefaultWriter,
writableStreamAbort,
writableStreamCloseQueuedOrInFlight,
writableStreamDefaultWriterCloseWithErrorPropagation,
writableStreamDefaultWriterRelease,
writableStreamDefaultWriterWriteWithRequest,
} = require('internal/webstreams/writablestream');
const { Buffer } = require('buffer');
const assert = require('internal/assert');
const kCancel = Symbol('kCancel');
const kClose = Symbol('kClose');
const kChunk = Symbol('kChunk');
const kError = Symbol('kError');
const kPull = Symbol('kPull');
const kRelease = Symbol('kRelease');
const kSkipThrow = Symbol('kSkipThrow');
let releasedError;
let releasingError;
let addAbortListener;
const userModuleRegExp = /^ {4}at (?:[^/\\(]+ \()(?!node:(.+):\d+:\d+\)$).*/gm;
function lazyReadableReleasedError() {
if (releasedError) {
return releasedError;
}
releasedError = new ERR_INVALID_STATE.TypeError('Reader released');
// Avoid V8 leak and remove userland stackstrace
releasedError.stack = SideEffectFreeRegExpPrototypeSymbolReplace(userModuleRegExp, releasedError.stack, '');
return releasedError;
}
function lazyReadableReleasingError() {
if (releasingError) {
return releasingError;
}
releasingError = new ERR_INVALID_STATE.TypeError('Releasing reader');
// Avoid V8 leak and remove userland stackstrace
releasingError.stack = SideEffectFreeRegExpPrototypeSymbolReplace(userModuleRegExp, releasingError.stack, '');
return releasingError;
}
/**
* @typedef {import('../abort_controller').AbortSignal} AbortSignal
* @typedef {import('./queuingstrategies').QueuingStrategy} QueuingStrategy
* @typedef {import('./queuingstrategies').QueuingStrategySize
* } QueuingStrategySize
* @typedef {import('./writablestream').WritableStream} WritableStream
*/
/**
* @typedef {ReadableStreamDefaultController | ReadableByteStreamController
* } ReadableStreamController
*/
/**
* @typedef {ReadableStreamDefaultReader | ReadableStreamBYOBReader
* } ReadableStreamReader
*/
/**
* @callback UnderlyingSourceStartCallback
* @param {ReadableStreamController} controller
* @returns { any | Promise<void> }
*/
/**
* @callback UnderlyingSourcePullCallback
* @param {ReadableStreamController} controller
* @returns { Promise<void> }
*/
/**
* @callback UnderlyingSourceCancelCallback
* @param {any} reason
* @returns { Promise<void> }
*/
/**
* @typedef {{
* readable: ReadableStream,
* writable: WritableStream,
* }} ReadableWritablePair
*/
/**
* @typedef {{
* preventClose? : boolean,
* preventAbort? : boolean,
* preventCancel? : boolean,
* signal? : AbortSignal,
* }} StreamPipeOptions
*/
/**
* @typedef {{
* start? : UnderlyingSourceStartCallback,
* pull? : UnderlyingSourcePullCallback,
* cancel? : UnderlyingSourceCancelCallback,
* type? : "bytes",
* autoAllocateChunkSize? : number
* }} UnderlyingSource
*/
class ReadableStream {
[kType] = 'ReadableStream';
/**
* @param {UnderlyingSource} [source]
* @param {QueuingStrategy} [strategy]
*/
constructor(source = kEmptyObject, strategy = kEmptyObject) {
markTransferMode(this, false, true);
// Internal construction (tee, transform streams, adapters, transfer):
// the caller sets up the controller, so every ReadableStream shares
// one hidden class and no per-instance prototype swap is needed.
if (source === kSkipThrow) {
this[kState] = createReadableStreamState();
return;
}
validateObject(source, 'source', kValidateObjectAllowObjects);
validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);
this[kState] = createReadableStreamState();
// The spec requires handling of the strategy first
// here. Specifically, if getting the size and
// highWaterMark from the strategy fail, that has
// to trigger a throw before getting the details
// from the source. So be sure to keep these in
// this order.
const size = strategy?.size;
const highWaterMark = strategy?.highWaterMark;
const type = source.type;
if (`${type}` === 'bytes') {
if (size !== undefined)
throw new ERR_INVALID_ARG_VALUE.RangeError('strategy.size', size);
setupReadableByteStreamControllerFromSource(
this,
source,
extractHighWaterMark(highWaterMark, 0));
} else {
if (type !== undefined)
throw new ERR_INVALID_ARG_VALUE('source.type', type);
setupReadableStreamDefaultControllerFromSource(
this,
source,
extractHighWaterMark(highWaterMark, 1),
extractSizeAlgorithm(size));
}
}
get [kIsDisturbed]() {
return this[kState].disturbed;
}
get [kIsErrored]() {
return this[kState].state === 'errored';
}
get [kIsReadable]() {
return this[kState].state === 'readable';
}
[kControllerErrorFunction](error) {
// Used by the internal stream interop (addAbortSignal). Historically
// only default controllers were wired here; byte stream controllers
// keep the previous no-op behavior.
const controller = this[kState].controller;
if (isReadableStreamDefaultController(controller))
controller.error(error);
}
// Used by the internal stream interop (end-of-stream). Materialized
// lazily since its settlement is derivable from the stream state; the
// settle sites in readableStreamClose/Error only touch the cache.
get [kIsClosedPromise]() {
let cache = this[kState].closedPromise;
if (cache === undefined) {
switch (this[kState].state) {
case 'readable':
cache = PromiseWithResolvers();
break;
case 'closed':
cache = resolvedRecord();
break;
default:
cache = rejectedHandledRecord(this[kState].storedError);
}
this[kState].closedPromise = cache;
}
return cache;
}
/**
* @readonly
* @type {boolean}
*/
get locked() {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
return isReadableStreamLocked(this);
}
static from(iterable) {
return readableStreamFromIterable(iterable);
}
/**
* @param {any} [reason]
* @returns { Promise<void> }
*/
cancel(reason = undefined) {
if (!isReadableStream(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStream'));
if (isReadableStreamLocked(this)) {
return PromiseReject(
new ERR_INVALID_STATE.TypeError('ReadableStream is locked'));
}
return readableStreamCancel(this, reason);
}
/**
* @param {{
* mode? : "byob"
* }} [options]
* @returns {ReadableStreamReader}
*/
getReader(options = kEmptyObject) {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
const mode = options?.mode;
if (mode === undefined)
// eslint-disable-next-line no-use-before-define
return new ReadableStreamDefaultReader(this);
if (`${mode}` !== 'byob')
throw new ERR_INVALID_ARG_VALUE('options.mode', mode);
// eslint-disable-next-line no-use-before-define
return new ReadableStreamBYOBReader(this);
}
/**
* @param {ReadableWritablePair} transform
* @param {StreamPipeOptions} [options]
* @returns {ReadableStream}
*/
pipeThrough(transform, options = kEmptyObject) {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
const readable = transform?.readable;
if (!isReadableStream(readable)) {
throw new ERR_INVALID_ARG_TYPE(
'transform.readable',
'ReadableStream',
readable);
}
const writable = transform?.writable;
if (!isWritableStream(writable)) {
throw new ERR_INVALID_ARG_TYPE(
'transform.writable',
'WritableStream',
writable);
}
// The web platform tests require that these be handled one at a
// time and in a specific order. options can be null or undefined.
validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
const preventAbort = options?.preventAbort;
const preventCancel = options?.preventCancel;
const preventClose = options?.preventClose;
const signal = options?.signal;
if (signal !== undefined) {
validateAbortSignal(signal, 'options.signal');
}
if (isReadableStreamLocked(this))
throw new ERR_INVALID_STATE.TypeError('The ReadableStream is locked');
if (isWritableStreamLocked(writable))
throw new ERR_INVALID_STATE.TypeError('The WritableStream is locked');
const promise = readableStreamPipeTo(
this,
writable,
!!preventClose,
!!preventAbort,
!!preventCancel,
signal);
setPromiseHandled(promise);
return readable;
}
/**
* @param {WritableStream} destination
* @param {StreamPipeOptions} [options]
* @returns {Promise<void>}
*/
pipeTo(destination, options = kEmptyObject) {
try {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
if (!isWritableStream(destination)) {
throw new ERR_INVALID_ARG_TYPE(
'transform.writable',
'WritableStream',
destination);
}
validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
const preventAbort = options?.preventAbort;
const preventCancel = options?.preventCancel;
const preventClose = options?.preventClose;
const signal = options?.signal;
if (signal !== undefined) {
validateAbortSignal(signal, 'options.signal');
}
if (isReadableStreamLocked(this))
throw new ERR_INVALID_STATE.TypeError('The ReadableStream is locked');
if (isWritableStreamLocked(destination))
throw new ERR_INVALID_STATE.TypeError('The WritableStream is locked');
return readableStreamPipeTo(
this,
destination,
!!preventClose,
!!preventAbort,
!!preventCancel,
signal);
} catch (error) {
return PromiseReject(error);
}
}
/**
* @returns {ReadableStream[]}
*/
tee() {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
return readableStreamTee(this, false);
}
/**
* @param {{
* preventCancel? : boolean,
* }} [options]
* @returns {AsyncIterable}
*/
values(options = kEmptyObject) {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
const preventCancel = !!(options?.preventCancel);
// eslint-disable-next-line no-use-before-define
const reader = new ReadableStreamDefaultReader(this);
// eslint-disable-next-line no-use-before-define
const state = new ReadableStreamAsyncIteratorReadRequest(reader, preventCancel);
// eslint-disable-next-line no-use-before-define
return new ReadableStreamAsyncIterator(state);
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
locked: this.locked,
state: this[kState].state,
supportsBYOB:
// eslint-disable-next-line no-use-before-define
this[kState].controller instanceof ReadableByteStreamController,
});
}
[kTransfer]() {
if (!isReadableStream(this))
throw new ERR_INVALID_THIS('ReadableStream');
const transfer = readableStreamTransferState(this);
if (this.locked) {
transfer.port1?.close();
transfer.port1 = undefined;
transfer.port2 = undefined;
throw new DOMException(
'Cannot transfer a locked ReadableStream',
'DataCloneError');
}
const {
writable,
promise,
} = lazyTransfer().newCrossRealmWritableSink(this, transfer.port1);
transfer.writable = writable;
transfer.promise = promise;
return {
data: { port: transfer.port2 },
deserializeInfo:
'internal/webstreams/readablestream:TransferredReadableStream',
};
}
[kTransferList]() {
const { port1, port2 } = new MessageChannel();
const transfer = readableStreamTransferState(this);
transfer.port1 = port1;
transfer.port2 = port2;
return [ port2 ];
}
[kDeserialize]({ port }) {
const transfer = lazyTransfer();
setupReadableStreamDefaultControllerFromSource(
this,
// The MessagePort is set to be referenced when reading.
// After two MessagePorts are closed, there is a problem with
// lingering promise not being properly resolved.
// https://github.com/nodejs/node/issues/51486
new transfer.CrossRealmTransformReadableSource(port, true),
0, defaultSizeAlgorithm);
}
}
ObjectDefineProperties(ReadableStream.prototype, {
[SymbolAsyncIterator]: {
__proto__: null,
configurable: true,
enumerable: false,
writable: true,
value: ReadableStream.prototype.values,
},
locked: kEnumerableProperty,
cancel: kEnumerableProperty,
getReader: kEnumerableProperty,
pipeThrough: kEnumerableProperty,
pipeTo: kEnumerableProperty,
tee: kEnumerableProperty,
values: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStream.name),
});
ObjectDefineProperties(ReadableStream, {
from: kEnumerableProperty,
});
function TransferredReadableStream() {
return new ReadableStream(kSkipThrow);
}
TransferredReadableStream.prototype[kDeserialize] = () => {};
class ReadableStreamBYOBRequest {
[kType] = 'ReadableStreamBYOBRequest';
constructor(skipThrowSymbol = undefined) {
if (skipThrowSymbol !== kSkipThrow) {
throw new ERR_ILLEGAL_CONSTRUCTOR();
}
}
/**
* @readonly
* @type {Uint8Array}
*/
get view() {
if (!isReadableStreamBYOBRequest(this))
throw new ERR_INVALID_THIS('ReadableStreamBYOBRequest');
return this[kState].view;
}
/**
* @param {number} bytesWritten
*/
respond(bytesWritten) {
if (!isReadableStreamBYOBRequest(this))
throw new ERR_INVALID_THIS('ReadableStreamBYOBRequest');
const {
view,
controller,
} = this[kState];
if (controller === undefined) {
throw new ERR_INVALID_STATE.TypeError(
'This BYOB request has been invalidated');
}
const viewByteLength = ArrayBufferViewGetByteLength(view);
const viewBuffer = ArrayBufferViewGetBuffer(view);
const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);
if (ArrayBufferPrototypeGetDetached(viewBuffer)) {
throw new ERR_INVALID_STATE.TypeError('Viewed ArrayBuffer is detached');
}
assert(viewByteLength > 0);
assert(viewBufferByteLength > 0);
readableByteStreamControllerRespond(controller, bytesWritten);
}
/**
* @param {ArrayBufferView} view
*/
respondWithNewView(view) {
if (!isReadableStreamBYOBRequest(this))
throw new ERR_INVALID_THIS('ReadableStreamBYOBRequest');
const {
controller,
} = this[kState];
if (controller === undefined) {
throw new ERR_INVALID_STATE.TypeError(
'This BYOB request has been invalidated');
}
validateBuffer(view, 'view');
if (ArrayBufferPrototypeGetDetached(view.buffer)) {
throw new ERR_INVALID_STATE.TypeError('Viewed ArrayBuffer is detached');
}
readableByteStreamControllerRespondWithNewView(controller, view);
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
view: this.view,
controller: this[kState].controller,
});
}
}
ObjectDefineProperties(ReadableStreamBYOBRequest.prototype, {
view: kEnumerableProperty,
respond: kEnumerableProperty,
respondWithNewView: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStreamBYOBRequest.name),
});
function createReadableStreamBYOBRequest(controller, view) {
const stream = new ReadableStreamBYOBRequest(kSkipThrow);
stream[kState] = {
controller,
view,
};
return stream;
}
// Per-iterator state. It doubles as the iterator's read request: at most
// one read is ever in flight (next() chains through `current`), and the
// request is consumed before the next read starts, so only its promise
// record changes per read.
class ReadableStreamAsyncIteratorReadRequest {
constructor(reader, preventCancel) {
this.reader = reader;
this.preventCancel = preventCancel;
this.done = false;
this.started = false;
this.current = undefined;
this.promise = undefined;
this.chainedNextSteps = undefined;
}
[kChunk](chunk) {
this.current = undefined;
this.promise.resolve({ value: chunk, done: false });
}
[kClose]() {
this.current = undefined;
this.done = true;
readableStreamReaderGenericRelease(this.reader);
this.promise.resolve({ value: undefined, done: true });
}
[kError](error) {
this.current = undefined;
this.done = true;
readableStreamReaderGenericRelease(this.reader);
this.promise.reject(error);
}
}
// next() is not an async function: it explicitly creates and returns a
// promise in the common case, so an async function would only add two
// promise allocations.
function readableStreamAsyncIteratorNextSteps(state) {
if (state.done)
return PromiseResolve({ value: undefined, done: true });
const reader = state.reader;
if (reader[kState].stream === undefined) {
return PromiseReject(
new ERR_INVALID_STATE.TypeError(
'The reader is not bound to a ReadableStream'));
}
const promise = PromiseWithResolvers();
state.promise = promise;
readableStreamDefaultReaderRead(reader, state);
return promise.promise;
}
// Not an async function either: the cancel path settles one microtask
// after the cancel promise, exactly as `await` would.
function readableStreamAsyncIteratorReturnSteps(state, value) {
const iterResult = { value, done: true };
if (state.done)
return PromiseResolve(iterResult);
state.done = true;
try {
const reader = state.reader;
if (reader[kState].stream === undefined) {
throw new ERR_INVALID_STATE.TypeError(
'The reader is not bound to a ReadableStream');
}
assert(!reader[kState].readRequests.length);
if (!state.preventCancel) {
const result = readableStreamReaderGenericCancel(reader, value);
readableStreamReaderGenericRelease(reader);
return PromisePrototypeThen(result, () => iterResult);
}
readableStreamReaderGenericRelease(reader);
return PromiseResolve(iterResult);
} catch (error) {
return PromiseReject(error);
}
}
// The methods live on a shared prototype, as for any WebIDL async iterator,
// rather than being created per iterator. Neither method may be an async
// function: `await` goes through Promise.prototype.then, which the streams
// WPTs patch.
class ReadableStreamAsyncIterator {
#state;
constructor(state) {
this.#state = state;
}
next() {
if (typeof this !== 'object' || this === null || !(#state in this)) {
return PromiseReject(
new ERR_INVALID_THIS('ReadableStreamAsyncIterator'));
}
const state = this.#state;
// If this is the first read, delay by one microtask
// to ensure that the controller has had an opportunity
// to properly start and perform the initial pull.
// TODO(@jasnell): The spec doesn't call this out so
// need to investigate if it's a bug in our impl or
// the spec.
if (!state.started) {
state.current = kResolvedPromise;
state.started = true;
}
if (state.current !== undefined) {
let steps = state.chainedNextSteps;
if (steps === undefined) {
steps = state.chainedNextSteps =
() => readableStreamAsyncIteratorNextSteps(state);
}
state.current = PromisePrototypeThen(state.current, steps, steps);
return state.current;
}
// No read is in flight. Mirror the buffered fast path of
// ReadableStreamDefaultReader.read(): when data is already queued
// in the controller, resolve immediately without allocating a
// read request. The result settles synchronously, so leaving
// state.current undefined matches the state the slow path reaches
// once its read request callbacks have settled.
const stream = state.reader[kState].stream;
if (!state.done && stream !== undefined &&
stream[kState].state === 'readable') {
const controller = stream[kState].controller;
if (isReadableStreamDefaultController(controller)) {
if (controller[kState].queue.length > 0) {
stream[kState].disturbed = true;
const chunk = dequeueValue(controller);
if (controller[kState].closeRequested &&
!controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// Reduced ShouldCallPull, as in the read() fast path.
readableStreamDefaultControllerPull(controller);
}
return PromiseResolve({ value: chunk, done: false });
}
} else if (controller[kState].queueTotalSize > 0) {
// Byte controller with buffered data: same shape as above via
// the queue-filled arm of the byte controller's pull steps.
stream[kState].disturbed = true;
return PromiseResolve({
value: readableByteStreamControllerDequeueChunk(controller),
done: false,
});
}
}
state.current = readableStreamAsyncIteratorNextSteps(state);
return state.current;
}
return(value) {
if (typeof this !== 'object' || this === null || !(#state in this)) {
return PromiseReject(
new ERR_INVALID_THIS('ReadableStreamAsyncIterator'));
}
const state = this.#state;
state.started = true;
state.current = state.current !== undefined ?
PromisePrototypeThen(
state.current,
() => readableStreamAsyncIteratorReturnSteps(state, value),
() => readableStreamAsyncIteratorReturnSteps(state, value)) :
readableStreamAsyncIteratorReturnSteps(state, value);
return state.current;
}
}
delete ReadableStreamAsyncIterator.prototype.constructor;
ObjectSetPrototypeOf(ReadableStreamAsyncIterator.prototype,
AsyncIteratorPrototype);
ObjectDefineProperties(ReadableStreamAsyncIterator.prototype, {
next: kEnumerableProperty,
return: kEnumerableProperty,
[SymbolToStringTag]:
getNonWritablePropertyDescriptor('ReadableStream AsyncIterator'),
});
class DefaultReadRequest {
constructor() {
this[kState] = PromiseWithResolvers();
}
[kChunk](value) {
this[kState].resolve({ value, done: false });
}
[kClose]() {
this[kState].resolve({ value: undefined, done: true });
}
[kError](error) {
this[kState].reject(error);
}
get promise() { return this[kState].promise; }
}
// Read (into) request record for the internal consumers (pipeTo, tee).
// The step functions are per-consumer closures, but the record is a class
// instance rather than an object literal: a literal with computed symbol
// keys is rebuilt through the runtime on every evaluation, which costs
// microseconds per pipe or tee (and per read in the byte tee's BYOB path).
class StepsReadRequest {
constructor(chunkSteps, closeSteps, errorSteps) {
this.chunkSteps = chunkSteps;
this.closeSteps = closeSteps;
this.errorSteps = errorSteps;
}
[kChunk](chunk) {
this.chunkSteps(chunk);
}
[kClose](chunk) {
this.closeSteps(chunk);
}
[kError](error) {
this.errorSteps(error);
}
}
class ReadIntoRequest {
constructor() {
this[kState] = PromiseWithResolvers();
}
[kChunk](value) {
this[kState].resolve({ value, done: false });
}
[kClose](value) {
this[kState].resolve({ value, done: true });
}
[kError](error) {
this[kState].reject(error);
}
get promise() { return this[kState].promise; }
}
class ReadableStreamDefaultReader {
[kType] = 'ReadableStreamDefaultReader';
/**
* @param {ReadableStream} stream
*/
constructor(stream) {
if (!isReadableStream(stream))
throw new ERR_INVALID_ARG_TYPE('stream', 'ReadableStream', stream);
this[kState] = {
// All fields are unconditionally assigned during setup.
readRequests: undefined,
stream: undefined,
close: undefined,
};
setupReadableStreamDefaultReader(this, stream);
}
/**
* @returns {Promise<{
* value : any,
* done : boolean
* }>}
*/
read() {
if (!isReadableStreamDefaultReader(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStreamDefaultReader'));
if (this[kState].stream === undefined) {
return PromiseReject(
new ERR_INVALID_STATE.TypeError(
'The reader is not attached to a stream'));
}
const stream = this[kState].stream;
const controller = stream[kState].controller;
// Fast path: if data is already buffered in the controller's queue,
// return a resolved promise immediately without creating a read request.
// This is spec-compliant because read() returns a Promise, and
// Promise.resolve() callbacks still run in the microtask queue.
if (stream[kState].state === 'readable') {
if (isReadableStreamDefaultController(controller)) {
if (controller[kState].queue.length > 0) {
stream[kState].disturbed = true;
const chunk = dequeueValue(controller);
if (controller[kState].closeRequested && !controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// ShouldCallPull reduced to the conditions not already
// established on this path: the state is readable and the
// queue was non-empty, so no read requests can be parked.
readableStreamDefaultControllerPull(controller);
}
return PromiseResolve({ value: chunk, done: false });
}
} else if (controller[kState].queueTotalSize > 0) {
// Byte controller with buffered data: mirror the queue-filled arm
// of its pull steps (which never consults pendingPullIntos) minus
// the read request.
stream[kState].disturbed = true;
return PromiseResolve({
value: readableByteStreamControllerDequeueChunk(controller),
done: false,
});
}
}
// Slow path: create request and go through normal flow
const readRequest = new DefaultReadRequest();
readableStreamDefaultReaderRead(this, readRequest);
return readRequest.promise;
}
releaseLock() {
if (!isReadableStreamDefaultReader(this))
throw new ERR_INVALID_THIS('ReadableStreamDefaultReader');
if (this[kState].stream === undefined)
return;
readableStreamDefaultReaderRelease(this);
}
/**
* @readonly
* @type {Promise<void>}
*/
get closed() {
if (!isReadableStreamDefaultReader(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStreamDefaultReader'));
return readerClosedPromise(this).promise;
}
/**
* @param {any} [reason]
* @returns {Promise<void>}
*/
cancel(reason = undefined) {
if (!isReadableStreamDefaultReader(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStreamDefaultReader'));
if (this[kState].stream === undefined) {
return PromiseReject(new ERR_INVALID_STATE.TypeError(
'The reader is not attached to a stream'));
}
return readableStreamReaderGenericCancel(this, reason);
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
stream: this[kState].stream,
readRequests: this[kState].readRequests.length,
close: readerClosedPromise(this).promise,
});
}
}
ObjectDefineProperties(ReadableStreamDefaultReader.prototype, {
closed: kEnumerableProperty,
read: kEnumerableProperty,
releaseLock: kEnumerableProperty,
cancel: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStreamDefaultReader.name),
});
class ReadableStreamBYOBReader {
[kType] = 'ReadableStreamBYOBReader';
/**
* @param {ReadableStream} stream
*/
constructor(stream) {
if (!isReadableStream(stream))
throw new ERR_INVALID_ARG_TYPE('stream', 'ReadableStream', stream);
this[kState] = {
// All fields are unconditionally assigned during setup.
stream: undefined,
readIntoRequests: undefined,
close: undefined,
};
setupReadableStreamBYOBReader(this, stream);
}
/**
* @param {ArrayBufferView} view
* @param {{
* min? : number
* }} [options]
* @returns {Promise<{
* value : ArrayBufferView,
* done : boolean,
* }>}
*/
read(view, options = kEmptyObject) {
// Returns the read request's promise directly, as the spec does,
// instead of adopting it through an async wrapper (an extra promise
// and two microtask hops per read); argument errors become rejections.
try {
return readableStreamBYOBReaderReadView(this, view, options);
} catch (error) {
return PromiseReject(error);
}
}
releaseLock() {
if (!isReadableStreamBYOBReader(this))
throw new ERR_INVALID_THIS('ReadableStreamBYOBReader');
if (this[kState].stream === undefined)
return;
readableStreamBYOBReaderRelease(this);
}
/**
* @readonly
* @type {Promise<void>}
*/
get closed() {
if (!isReadableStreamBYOBReader(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStreamBYOBReader'));
return readerClosedPromise(this).promise;
}
/**
* @param {any} [reason]
* @returns {Promise<void>}
*/
cancel(reason = undefined) {
if (!isReadableStreamBYOBReader(this))
return PromiseReject(new ERR_INVALID_THIS('ReadableStreamBYOBReader'));
if (this[kState].stream === undefined) {
return PromiseReject(new ERR_INVALID_STATE.TypeError(
'The reader is not attached to a stream'));
}
return readableStreamReaderGenericCancel(this, reason);
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
stream: this[kState].stream,
readIntoRequests: this[kState].readIntoRequests.length,
close: readerClosedPromise(this).promise,
});
}
}
ObjectDefineProperties(ReadableStreamBYOBReader.prototype, {
closed: kEnumerableProperty,
read: kEnumerableProperty,
releaseLock: kEnumerableProperty,
cancel: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStreamBYOBReader.name),
});
class ReadableStreamDefaultController {
[kType] = 'ReadableStreamDefaultController';
[kState];
constructor(skipThrowSymbol = undefined) {
if (skipThrowSymbol !== kSkipThrow) {
throw new ERR_ILLEGAL_CONSTRUCTOR();
}
}
/**
* @readonly
* @type {number}
*/
get desiredSize() {
return readableStreamDefaultControllerGetDesiredSize(this);
}
close() {
if (!readableStreamDefaultControllerCanCloseOrEnqueue(this))
throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
readableStreamDefaultControllerClose(this);
}
/**
* @param {any} [chunk]
*/
enqueue(chunk = undefined) {
if (!readableStreamDefaultControllerCanCloseOrEnqueue(this))
throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
readableStreamDefaultControllerEnqueue(this, chunk);
}
/**
* @param {any} [error]
*/
error(error = undefined) {
readableStreamDefaultControllerError(this, error);
}
[kCancel](reason) {
return readableStreamDefaultControllerCancelSteps(this, reason);
}
[kPull](readRequest) {
readableStreamDefaultControllerPullSteps(this, readRequest);
}
[kRelease]() {}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], { });
}
}
ObjectDefineProperties(ReadableStreamDefaultController.prototype, {
desiredSize: kEnumerableProperty,
close: kEnumerableProperty,
enqueue: kEnumerableProperty,
error: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStreamDefaultController.name),
});
class ReadableByteStreamController {
[kType] = 'ReadableByteStreamController';
[kState];
constructor(skipThrowSymbol = undefined) {
if (skipThrowSymbol !== kSkipThrow) {
throw new ERR_ILLEGAL_CONSTRUCTOR();
}
}
/**
* @readonly
* @type {ReadableStreamBYOBRequest}
*/
get byobRequest() {
if (!isReadableByteStreamController(this))
throw new ERR_INVALID_THIS('ReadableByteStreamController');
if (this[kState].byobRequest === null &&
this[kState].pendingPullIntos.length) {
const {
buffer,
byteOffset,
bytesFilled,
byteLength,
} = this[kState].pendingPullIntos.peek();
const view =
new Uint8Array(
buffer,
byteOffset + bytesFilled,
byteLength - bytesFilled);
this[kState].byobRequest = createReadableStreamBYOBRequest(this, view);
}
return this[kState].byobRequest;
}
/**
* @readonly
* @type {number}
*/
get desiredSize() {
if (!isReadableByteStreamController(this))
throw new ERR_INVALID_THIS('ReadableByteStreamController');
return readableByteStreamControllerGetDesiredSize(this);
}
close() {
if (!isReadableByteStreamController(this))
throw new ERR_INVALID_THIS('ReadableByteStreamController');
if (this[kState].closeRequested)
throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
if (this[kState].stream[kState].state !== 'readable')
throw new ERR_INVALID_STATE.TypeError('ReadableStream is already closed');
readableByteStreamControllerClose(this);
}
/**
* @param {ArrayBufferView} chunk
*/
enqueue(chunk) {
if (!isReadableByteStreamController(this))
throw new ERR_INVALID_THIS('ReadableByteStreamController');
validateBuffer(chunk);
const chunkByteLength = ArrayBufferViewGetByteLength(chunk);
const chunkBuffer = ArrayBufferViewGetBuffer(chunk);
if (isSharedArrayBuffer(chunkBuffer)) {
throw new ERR_INVALID_ARG_VALUE(
'chunk',
chunk,
'must not be backed by a SharedArrayBuffer',
);
}
const chunkBufferByteLength = ArrayBufferPrototypeGetByteLength(chunkBuffer);
if (chunkByteLength === 0 || chunkBufferByteLength === 0) {
throw new ERR_INVALID_STATE.TypeError(
'chunk ArrayBuffer is zero-length or detached');
}
if (this[kState].closeRequested)
throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
if (this[kState].stream[kState].state !== 'readable')
throw new ERR_INVALID_STATE.TypeError('ReadableStream is already closed');
readableByteStreamControllerEnqueue(this, chunk);
}
/**
* @param {any} [error]
*/
error(error = undefined) {
if (!isReadableByteStreamController(this))
throw new ERR_INVALID_THIS('ReadableByteStreamController');
readableByteStreamControllerError(this, error);
}
[kCancel](reason) {
return readableByteStreamControllerCancelSteps(this, reason);
}
[kPull](readRequest) {
readableByteStreamControllerPullSteps(this, readRequest);
}
[kRelease]() {
const {
pendingPullIntos,
} = this[kState];
if (pendingPullIntos.length > 0) {
const firstPendingPullInto = pendingPullIntos.peek();
firstPendingPullInto.type = 'none';
const queue = new Queue();
queue.push(firstPendingPullInto);
this[kState].pendingPullIntos = queue;
}
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], { });
}
}
ObjectDefineProperties(ReadableByteStreamController.prototype, {
byobRequest: kEnumerableProperty,
desiredSize: kEnumerableProperty,
close: kEnumerableProperty,
enqueue: kEnumerableProperty,
error: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
});
function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) {
const stream = new ReadableStream(kSkipThrow);
setupReadableStreamDefaultController(
stream,
new ReadableStreamDefaultController(kSkipThrow),
start,
pull,
cancel,
highWaterMark,
size);
return stream;
}
function createReadableByteStream(start, pull, cancel) {
const stream = new ReadableStream(kSkipThrow);
setupReadableByteStreamController(
stream,
new ReadableByteStreamController(kSkipThrow),
start,
pull,
cancel,
0,
undefined);
return stream;
}
const isReadableStream =
isBrandCheck('ReadableStream');
const isReadableByteStreamController =
isBrandCheck('ReadableByteStreamController');
const isReadableStreamDefaultController =
isBrandCheck('ReadableStreamDefaultController');
const isReadableStreamBYOBRequest =
isBrandCheck('ReadableStreamBYOBRequest');
const isReadableStreamDefaultReader =
isBrandCheck('ReadableStreamDefaultReader');
const isReadableStreamBYOBReader =
isBrandCheck('ReadableStreamBYOBReader');
// ---- ReadableStream Implementation
// The state record classes null out their prototype chain to keep
// Object.prototype out of property lookups; every field ever assigned
// is declared so the shape never transitions.
class ReadableStreamTransferState {
writable = undefined;
port1 = undefined;
port2 = undefined;
promise = undefined;
}
ObjectSetPrototypeOf(ReadableStreamTransferState.prototype, null);
class ReadableStreamState {
closedPromise = undefined;
disturbed = false;
reader = undefined;
state = 'readable';
storedError = undefined;
controller = undefined;
// Materialized on transfer, see readableStreamTransferState().
transfer = undefined;
}
ObjectSetPrototypeOf(ReadableStreamState.prototype, null);
function readableStreamTransferState(stream) {
return stream[kState].transfer ??= new ReadableStreamTransferState();
}
function createReadableStreamState() {
return new ReadableStreamState();
}
function readableStreamFromIterable(iterable) {
let stream;
const iteratorGetter = iterable[SymbolAsyncIterator] ?? iterable[SymbolIterator];
if (iteratorGetter == null || typeof iteratorGetter !== 'function') {
throw new ERR_ARG_NOT_ITERABLE(iterable);
}
const iterator = FunctionPrototypeCall(iteratorGetter, iterable);
if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) {
throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
}
// Per GetIteratorDirect, the next method is looked up once.
const nextMethod = iterator.next;
const startAlgorithm = nonOpCallback;
// Callback-style pull: the reaction steps are reused across chunks and
// completion is delivered to the controller's cached pull reactions
// (the kParkedAlgorithmResult contract). One pull runs at a time, so a
// single slot carries a non-thenable next() result between steps.
let pendingIterResult;
function rejectPull(error) {
readableStreamDefaultControllerError(stream[kState].controller, error);
}
function processIterResult(iterResult) {
const controller = stream[kState].controller;
if (typeof iterResult !== 'object' || iterResult === null) {
rejectPull(new ERR_INVALID_STATE.TypeError(
'The promise returned by the iterator.next() method must fulfill with an object'));
return;
}
try {
if (iterResult.done) {
readableStreamDefaultControllerClose(controller);
} else {
const value = iterResult.value;
if (value !== null &&
(typeof value === 'object' || typeof value === 'function')) {
// Adopted like `await iterResult.value`, keeping the observable
// .then lookup on plain objects.
PromisePrototypeThen(PromiseResolve(value), enqueueValue, rejectPull);
return;
}
readableStreamDefaultControllerEnqueue(controller, value);
}
} catch (error) {
rejectPull(error);
return;
}
// pullFulfilled exists: the controller creates it before the pull.
controller[kState].pullFulfilled();
}
function enqueueValue(value) {
const controller = stream[kState].controller;
try {
readableStreamDefaultControllerEnqueue(controller, value);
} catch (error) {
rejectPull(error);
return;
}
controller[kState].pullFulfilled();
}
function processPendingIterResult() {
const iterResult = pendingIterResult;
pendingIterResult = undefined;
processIterResult(iterResult);
}
function pullAlgorithm() {
let nextResult;
try {
nextResult = FunctionPrototypeCall(nextMethod, iterator);
} catch (error) {
return PromiseReject(error);
}
if (nextResult !== null &&
(typeof nextResult === 'object' || typeof nextResult === 'function')) {
// Mirrors `await iterator.next()`: processIterResult runs at the
// microtask position the await resumed.
PromisePrototypeThen(
PromiseResolve(nextResult), processIterResult, rejectPull);
return kParkedAlgorithmResult;
}
// A non-thenable next() result fails validation a microtask later.
pendingIterResult = nextResult;
PromisePrototypeThen(kResolvedPromise, processPendingIterResult);
return kParkedAlgorithmResult;
}
async function cancelAlgorithm(reason) {
const returnMethod = iterator.return;
if (returnMethod === undefined) {
return;
}
const iterResult = await FunctionPrototypeCall(returnMethod, iterator, reason);
if (typeof iterResult !== 'object' || iterResult === null) {
throw new ERR_INVALID_STATE.TypeError(
'The promise returned by the iterator.return() method must fulfill with an object');
}
}
stream = createReadableStream(
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
0,
);
return stream;
}
// Duck-types the lazily materialized [[closedPromise]] record of a reader
// or writer that only pipeTo or tee holds a reference to. Settling the
// record enqueues the watcher at the microtask position a reaction on the
// promise would have had, without materializing the promise; the shared
// pending promise satisfies the probes on the erroring/release paths.
// The reaction is registered in the async context captured at install
// time, where the spec registers it, rather than in the context of
// whoever settles the record.
class ClosedPromiseHook {
constructor(onResolved, onRejected) {
this.promise = kPendingPromise;
this.onResolved = onResolved;
this.onRejected = onRejected;
this.contextFrame = AsyncContextFrame.current();
}
resolve() {
if (this.onResolved === undefined)
return;
const prior = AsyncContextFrame.exchange(this.contextFrame);
PromisePrototypeThen(kResolvedPromise, this.onResolved);
AsyncContextFrame.set(prior);
}
reject(error) {
const onRejected = this.onRejected;
const prior = AsyncContextFrame.exchange(this.contextFrame);
PromisePrototypeThen(kResolvedPromise, () => onRejected(error));
AsyncContextFrame.set(prior);
}
}
// Installs an error watcher as the [[closedPromise]] record of a reader
// that only its caller holds. A reader of an already errored stream would
// have observed a rejected promise, so its watcher is enqueued right away.
function watchReaderErrored(reader, onRejected) {
const stream = reader[kState].stream;
if (stream[kState].state === 'errored') {
const error = stream[kState].storedError;
PromisePrototypeThen(kResolvedPromise, () => onRejected(error));
return;
}
reader[kState].close = new ClosedPromiseHook(undefined, onRejected);
}
function readableStreamPipeTo(
source,
dest,
preventClose,
preventAbort,
preventCancel,
signal) {
let reader;
let writer;
let disposable;
// Both of these can throw synchronously. We want to capture
// the error and return a rejected promise instead.
try {
reader = new ReadableStreamDefaultReader(source);
writer = new WritableStreamDefaultWriter(dest);
} catch (error) {
return PromiseReject(error);
}
source[kState].disturbed = true;
let shuttingDown = false;
if (signal !== undefined) {
try {
validateAbortSignal(signal, 'options.signal');
} catch (error) {
return PromiseReject(error);
}
}
const promise = PromiseWithResolvers();
// One shared write request tracks every chunk written to the
// destination, instead of a { promise, resolve, reject } record per
// write. `stall` is armed by waitForPendingWrites() during shutdown;
// `failed`/`failure` latch a write that could not proceed.
const writeTracker = {
// Non-undefined: queue entries are discriminated from kNilRequest
// by `promise === undefined`.
promise: null,
pending: 0,
failed: false,
failure: undefined,
stall: null,
resolve() {
if (--this.pending === 0 && this.stall !== null) {
const stall = this.stall;
this.stall = null;
if (this.failed)
stall.reject(this.failure);
else
stall.resolve();
}
},
reject(error) {
this.pending--;
this.failed = true;
this.failure = error;
if (this.stall !== null) {
const stall = this.stall;
this.stall = null;
stall.reject(error);
}
},
};
// The error here can be undefined. The rejected arg
// tells us that the promise must be rejected even
// when error is undefine.
function finalize(rejected, error) {
// The pipe is the only observer of the reader's and writer's promise
// records (including the ready hook installed by parkOnReady), and
// it is done with them: dropping them lets release skip the
// pending-promise probes and the rejections nothing would handle.
writer[kState].ready = undefined;
writer[kState].close = undefined;
reader[kState].close = undefined;
writableStreamDefaultWriterRelease(writer);
readableStreamReaderGenericRelease(reader);
if (signal !== undefined)
disposable?.[SymbolDispose]();
if (rejected)
promise.reject(error);
else
promise.resolve();
}
function waitForPendingWrites() {
if (writeTracker.pending === 0) {
return writeTracker.failed ?
PromiseReject(writeTracker.failure) :
PromiseResolve();
}
writeTracker.stall = PromiseWithResolvers();
return writeTracker.stall.promise;
}
function shutdownWithAnAction(action, rejected, originalError) {
if (shuttingDown) return;
shuttingDown = true;
if (dest[kState].state === 'writable' &&
!writableStreamCloseQueuedOrInFlight(dest)) {
PromisePrototypeThen(
waitForPendingWrites(),
complete,
(error) => finalize(true, error));
return;
}
complete();
function complete() {
PromisePrototypeThen(
action(),
() => finalize(rejected, originalError),
(error) => finalize(true, error));
}
}
function shutdown(rejected, error) {
if (shuttingDown) return;
shuttingDown = true;
if (dest[kState].state === 'writable' &&
!writableStreamCloseQueuedOrInFlight(dest)) {
PromisePrototypeThen(
waitForPendingWrites(),
() => finalize(rejected, error),
(error) => finalize(true, error));
return;
}
finalize(rejected, error);
}
function abortAlgorithm() {
let error;
if (signal.reason instanceof AbortError) {
// Cannot use the AbortError class here. It must be a DOMException.
error = new DOMException(signal.reason.message, 'AbortError');
} else {
error = signal.reason;
}
const actions = [];
if (!preventAbort) {
ArrayPrototypePush(
actions,
() => {
if (dest[kState].state === 'writable')
return writableStreamAbort(dest, error);
return PromiseResolve();
});
}
if (!preventCancel) {
ArrayPrototypePush(
actions,
() => {
if (source[kState].state === 'readable')
return readableStreamCancel(source, error);
return PromiseResolve();
});
}
shutdownWithAnAction(
() => SafePromiseAll(actions, (action) => action()),
true,
error);
}
// The pump loop is callback-driven to avoid per-iteration promise
// allocations. At most one read is in flight at a time, so one read
// request and one forwarding function are reused for every chunk;
// the chunk travels through `pendingChunk`.
let pendingChunk;
let readRequest;
let readyHook;
// Ready promise rejection is handled by the destination-errored
// watcher.
function ignoreReadyRejection() {}
// Parks the pump on the destination's backpressure by installing a
// record that duck-types the writer's lazily-materialized
// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
// when backpressure clears (after publishing the new backpressure
// state), which re-enters the pump directly instead of rotating a
// fresh promise record plus reaction per flip. The pipe holds the only
// reference to the writer, so the record is never observable as a real
// ready promise; the erroring/release paths probe `promise` via
// isPromisePending() and call `reject`, so it carries the shared
// pending promise and a no-op reject.
function parkOnReady() {
readyHook ??= {
promise: kPendingPromise,
resolve: pump,
reject: ignoreReadyRejection,
};
writer[kState].ready = readyHook;
}
function forwardChunk() {
const chunk = pendingChunk;
pendingChunk = undefined;
if (writer[kState].stream === undefined) return;
writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
pump();
}
function pump() {
if (shuttingDown) return;
if (dest[kState].backpressure) {
parkOnReady();
return;
}
const controller = source[kState].controller;
// Fast path: batch reads when data is buffered in a default controller.
// This avoids parking read requests and reduces promise allocation
// overhead.
if (source[kState].state === 'readable' &&
isReadableStreamDefaultController(controller) &&
controller[kState].queue.length > 0) {
while (controller[kState].queue.length > 0) {
if (shuttingDown) return;
const chunk = dequeueValue(controller);
if (controller[kState].closeRequested && !controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(source);
}
// Write the chunk - we're already in a separate microtask from enqueue
// because we awaited the writer ready promise above.
writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
// Check backpressure after each write
if (dest[kState].backpressure) {
// Backpressure - stop batch and wait for ready
break;
} else if (dest[kState].state !== 'writable' || writableStreamCloseQueuedOrInFlight(dest)) {
// Closing or erroring - stop batch and wait for shutdown
break;
}
}
// Trigger pull if needed after batch
readableStreamDefaultControllerCallPullIfNeeded(controller);
// Check if stream closed during batch
if (source[kState].state === 'closed') {
return;
}
// Park on backpressure directly: the ready hook resumes the pump
// when a completed write clears it.
if (dest[kState].backpressure) {
parkOnReady();
return;
}
// Yield to the microtask queue between batches so completed-write
// reactions and events/signals fire; a shared resolved promise
// enqueues the continuation at the same position as queueMicrotask
// without the per-batch scheduling overhead.
PromisePrototypeThen(kResolvedPromise, pump);
return;
}
// Slow path: park a lazily materialized read request. Close and
// error are handled by the source watchers.
readRequest ??= new StepsReadRequest(
(chunk) => {
// Per spec, pipeTo must queue a microtask for the write to avoid
// synchronous write during enqueue(). See WHATWG Streams spec
// "ReadableStreamPipeTo" step 15's "chunk steps".
pendingChunk = chunk;
PromisePrototypeThen(kResolvedPromise, forwardChunk);
},
nonOpCallback,
nonOpCallback);
readableStreamDefaultReaderRead(reader, readRequest);
}
if (signal !== undefined) {
if (signal.aborted) {
abortAlgorithm();
return promise.promise;
}
addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
disposable = addAbortListener(signal, abortAlgorithm);
}
pump();
function onSourceErrored(error) {
if (!preventAbort) {
return shutdownWithAnAction(
() => writableStreamAbort(dest, error),
true,
error);
}
shutdown(true, error);
}
function onSourceClosed() {
if (!preventClose) {
return shutdownWithAnAction(
() => writableStreamDefaultWriterCloseWithErrorPropagation(writer));
}
shutdown();
}
function onDestErrored(error) {
if (!preventCancel) {
return shutdownWithAnAction(
() => readableStreamCancel(source, error),
true,
error);
}
shutdown(true, error);
}
// The spec installs the source-errored watcher before the dest-errored
// one and the source-closed watcher last; a source that is already
// errored is handled before the dest watcher is installed, and an
// already-closed source after it, as before. The pipe holds the only
// references to its reader and writer, so instead of reacting to their
// [[closedPromise]] the watchers are installed as the records
// themselves (see ClosedPromiseHook).
if (source[kState].state === 'errored') {
onSourceErrored(source[kState].storedError);
} else if (source[kState].state !== 'closed') {
reader[kState].close =
new ClosedPromiseHook(onSourceClosed, onSourceErrored);
}
if (dest[kState].state === 'errored') {
onDestErrored(dest[kState].storedError);
} else {
writer[kState].close = new ClosedPromiseHook(undefined, onDestErrored);
}
if (source[kState].state === 'closed')
onSourceClosed();
if (writableStreamCloseQueuedOrInFlight(dest) ||
dest[kState].state === 'closed') {
const error = new ERR_INVALID_STATE.TypeError(
'Destination WritableStream is closed');
if (!preventCancel) {
shutdownWithAnAction(
() => readableStreamCancel(source, error), true, error);
} else {
shutdown(true, error);
}
}
return promise.promise;
}
function readableStreamTee(stream, cloneForBranch2) {
if (isReadableByteStreamController(stream[kState].controller)) {
return readableByteStreamTee(stream);
}
return readableStreamDefaultTee(stream, cloneForBranch2);
}
function readableStreamDefaultTee(stream, cloneForBranch2) {
const reader = new ReadableStreamDefaultReader(stream);
let reading = false;
let canceled1 = false;
let canceled2 = false;
let reason1;
let reason2;
let branch1;
let branch2;
// The spec's cancelPromise is materialized by the first branch cancel;
// until then its settlement is tracked by `cancelSettled`, so a tee
// whose branches are never canceled allocates no promise record for it.
let cancelPromise;
let cancelSettled = false;
function settleCancelPromise() {
if (cancelPromise !== undefined)
cancelPromise.resolve();
else
cancelSettled = true;
}
function cancelPromiseRecord() {
if (cancelPromise === undefined) {
cancelPromise = PromiseWithResolvers();
if (cancelSettled)
cancelPromise.resolve();
}
return cancelPromise;
}
// At most one read is ever in flight (`reading` guards pullAlgorithm),
// so one read request object and one forwarding microtask function are
// reused for every chunk; the chunk travels through `pendingChunk`.
// The request is materialized lazily on the first pull so that tee()
// itself stays allocation-light.
let pendingChunk;
let readRequest;
function forwardChunk() {
reading = false;
const value1 = pendingChunk;
let value2 = pendingChunk;
pendingChunk = undefined;
if (!canceled2 && cloneForBranch2) {
value2 = structuredClone(value2);
}
if (!canceled1) {
readableStreamDefaultControllerEnqueue(
branch1[kState].controller,
value1);
}
if (!canceled2) {
readableStreamDefaultControllerEnqueue(
branch2[kState].controller,
value2);
}
}
// A non-thenable pull result reaches the controller's pull-fulfilled
// step at the same microtask position as the async wrapper's promise
// did, without the implicit promise per pull.
function pullAlgorithm() {
if (reading) return;
reading = true;
readRequest ??= new StepsReadRequest(
(value) => {
// The microtask is required by the spec (ReadableStreamTee's
// "chunk steps" queue one).
pendingChunk = value;
PromisePrototypeThen(kResolvedPromise, forwardChunk);
},
() => {
// The `process.nextTick()` is not part of the spec.
// This approach was needed to avoid a race condition working with esm
// Further information, see: https://github.com/nodejs/node/issues/39758
process.nextTick(() => {
reading = false;
if (!canceled1)
readableStreamDefaultControllerClose(branch1[kState].controller);
if (!canceled2)
readableStreamDefaultControllerClose(branch2[kState].controller);
if (!canceled1 || !canceled2)
settleCancelPromise();
});
},
() => {
reading = false;
});
readableStreamDefaultReaderRead(reader, readRequest);
}
function cancel1Algorithm(reason) {
canceled1 = true;
reason1 = reason;
const record = cancelPromiseRecord();
if (canceled2) {
const compositeReason = [reason1, reason2];
record.resolve(readableStreamCancel(stream, compositeReason));
}
return record.promise;
}
function cancel2Algorithm(reason) {
canceled2 = true;
reason2 = reason;
const record = cancelPromiseRecord();
if (canceled1) {
const compositeReason = [reason1, reason2];
record.resolve(readableStreamCancel(stream, compositeReason));
}
return record.promise;
}
branch1 =
createReadableStream(nonOpCallback, pullAlgorithm, cancel1Algorithm);
branch2 =
createReadableStream(nonOpCallback, pullAlgorithm, cancel2Algorithm);
// The tee holds the only reference to the reader, so the error watcher
// is installed as its [[closedPromise]] record (see ClosedPromiseHook).
watchReaderErrored(reader, (error) => {
readableStreamDefaultControllerError(branch1[kState].controller, error);
readableStreamDefaultControllerError(branch2[kState].controller, error);
if (!canceled1 || !canceled2)
settleCancelPromise();
});
return [branch1, branch2];
}
function readableByteStreamTee(stream) {
assert(isReadableStream(stream));
assert(isReadableByteStreamController(stream[kState].controller));
let reader = new ReadableStreamDefaultReader(stream);
let reading = false;
let readAgainForBranch1 = false;
let readAgainForBranch2 = false;
let canceled1 = false;
let canceled2 = false;
let reason1;
let reason2;
let branch1;
let branch2;
// See readableStreamDefaultTee.
let cancelDeferred;
let cancelSettled = false;
function settleCancelDeferred() {
if (cancelDeferred !== undefined)
cancelDeferred.resolve();
else
cancelSettled = true;
}
function cancelDeferredRecord() {
if (cancelDeferred === undefined) {
cancelDeferred = PromiseWithResolvers();
if (cancelSettled)
cancelDeferred.resolve();
}
return cancelDeferred;
}
// The tee holds the only reference to each reader it creates, so the
// error watcher is installed as the reader's [[closedPromise]] record
// (see ClosedPromiseHook); releasing a reader rejects it like the
// promise, and the guard below ignores a reader that was swapped out.
function forwardReaderError(thisReader) {
watchReaderErrored(thisReader, (error) => {
if (thisReader !== reader) {
return;
}
readableStreamDefaultControllerError(branch1[kState].controller, error);
readableStreamDefaultControllerError(branch2[kState].controller, error);
if (!canceled1 || !canceled2) {
settleCancelDeferred();
}
});
}
// As in readableStreamDefaultTee, only one read is ever in flight, so
// the default-reader read request and its forwarding microtask are
// shared across all chunks.
let pendingChunk;
function forwardChunk() {
readAgainForBranch1 = false;
readAgainForBranch2 = false;
const chunk1 = pendingChunk;
let chunk2 = pendingChunk;
pendingChunk = undefined;
if (!canceled1 && !canceled2) {
try {
chunk2 = cloneAsUint8Array(chunk1);
} catch (error) {
readableByteStreamControllerError(
branch1[kState].controller,
error,
);
readableByteStreamControllerError(
branch2[kState].controller,
error,
);
cancelDeferredRecord().resolve(readableStreamCancel(stream, error));
return;
}
}
if (!canceled1) {
readableByteStreamControllerEnqueue(
branch1[kState].controller,
chunk1,
);
}
if (!canceled2) {
readableByteStreamControllerEnqueue(
branch2[kState].controller,
chunk2,
);
}
reading = false;
if (readAgainForBranch1) {
pull1Algorithm();
} else if (readAgainForBranch2) {
pull2Algorithm();
}
}
let defaultReadRequest;
function pullWithDefaultReader() {
if (isReadableStreamBYOBReader(reader)) {
readableStreamBYOBReaderRelease(reader);
reader = new ReadableStreamDefaultReader(stream);
forwardReaderError(reader);
}
defaultReadRequest ??= new StepsReadRequest(
(chunk) => {
pendingChunk = chunk;
PromisePrototypeThen(kResolvedPromise, forwardChunk);
},
() => {
reading = false;
if (!canceled1) {
readableByteStreamControllerClose(branch1[kState].controller);
}
if (!canceled2) {
readableByteStreamControllerClose(branch2[kState].controller);
}
if (branch1[kState].controller[kState].pendingPullIntos.length > 0) {
readableByteStreamControllerRespond(branch1[kState].controller, 0);
}
if (branch2[kState].controller[kState].pendingPullIntos.length > 0) {
readableByteStreamControllerRespond(branch2[kState].controller, 0);
}
if (!canceled1 || !canceled2) {
settleCancelDeferred();
}
},
() => {
reading = false;
});
readableStreamDefaultReaderRead(reader, defaultReadRequest);
}
function pullWithBYOBReader(view, forBranch2) {
if (isReadableStreamDefaultReader(reader)) {
readableStreamDefaultReaderRelease(reader);
reader = new ReadableStreamBYOBReader(stream);
forwardReaderError(reader);
}
const byobBranch = forBranch2 === true ? branch2 : branch1;
const otherBranch = forBranch2 === false ? branch2 : branch1;
const readIntoRequest = new StepsReadRequest(
(chunk) => {
queueMicrotask(() => {
readAgainForBranch1 = false;
readAgainForBranch2 = false;
const byobCanceled = forBranch2 === true ? canceled2 : canceled1;
const otherCanceled = forBranch2 === false ? canceled2 : canceled1;
if (!otherCanceled) {
let clonedChunk;
try {
clonedChunk = cloneAsUint8Array(chunk);
} catch (error) {
readableByteStreamControllerError(
byobBranch[kState].controller,
error,
);
readableByteStreamControllerError(
otherBranch[kState].controller,
error,
);
cancelDeferredRecord().resolve(readableStreamCancel(stream, error));
return;
}
if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk,
);
}
readableByteStreamControllerEnqueue(
otherBranch[kState].controller,
clonedChunk,
);
} else if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk,
);
}
reading = false;
if (readAgainForBranch1) {
pull1Algorithm();
} else if (readAgainForBranch2) {
pull2Algorithm();
}
});
},
(chunk) => {
reading = false;
const byobCanceled = forBranch2 === true ? canceled2 : canceled1;
const otherCanceled = forBranch2 === false ? canceled2 : canceled1;
if (!byobCanceled) {
readableByteStreamControllerClose(byobBranch[kState].controller);
}
if (!otherCanceled) {
readableByteStreamControllerClose(otherBranch[kState].controller);
}
if (chunk !== undefined) {
if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk,
);
}
if (
!otherCanceled &&
otherBranch[kState].controller[kState].pendingPullIntos.length > 0
) {
readableByteStreamControllerRespond(
otherBranch[kState].controller,
0,
);
}
}
if (!byobCanceled || !otherCanceled) {
settleCancelDeferred();
}
},
() => {
reading = false;
});
readableStreamBYOBReaderRead(reader, view, 1, readIntoRequest);
}
function pull1Algorithm() {
if (reading) {
readAgainForBranch1 = true;
return;
}
reading = true;
const byobRequest = branch1[kState].controller.byobRequest;
if (byobRequest === null) {
pullWithDefaultReader();
} else {
pullWithBYOBReader(byobRequest[kState].view, false);
}
}
function pull2Algorithm() {
if (reading) {
readAgainForBranch2 = true;
return;
}
reading = true;
const byobRequest = branch2[kState].controller.byobRequest;
if (byobRequest === null) {
pullWithDefaultReader();
} else {
pullWithBYOBReader(byobRequest[kState].view, true);
}
}
function cancel1Algorithm(reason) {
canceled1 = true;
reason1 = reason;
const record = cancelDeferredRecord();
if (canceled2) {
record.resolve(readableStreamCancel(stream, [reason1, reason2]));
}
return record.promise;
}
function cancel2Algorithm(reason) {
canceled2 = true;
reason2 = reason;
const record = cancelDeferredRecord();
if (canceled1) {
record.resolve(readableStreamCancel(stream, [reason1, reason2]));
}
return record.promise;
}
branch1 =
createReadableByteStream(nonOpCallback, pull1Algorithm, cancel1Algorithm);
branch2 =
createReadableByteStream(nonOpCallback, pull2Algorithm, cancel2Algorithm);
forwardReaderError(reader);
return [branch1, branch2];
}
function readableByteStreamControllerConvertPullIntoDescriptor(desc) {
const {
buffer,
bytesFilled,
byteLength,
byteOffset,
ctor,
elementSize,
} = desc;
if (bytesFilled > byteLength)
throw new ERR_INVALID_STATE.RangeError('The buffer size is invalid');
assert(!(bytesFilled % elementSize));
const transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
if (ctor === Buffer) {
return Buffer.from(transferredBuffer, byteOffset, bytesFilled / elementSize);
}
return new ctor(transferredBuffer, byteOffset, bytesFilled / elementSize);
}
function isReadableStreamLocked(stream) {
return stream[kState].reader !== undefined;
}
function readableStreamCancel(stream, reason) {
stream[kState].disturbed = true;
switch (stream[kState].state) {
case 'closed':
return PromiseResolve();
case 'errored':
return PromiseReject(stream[kState].storedError);
}
readableStreamClose(stream);
const {
reader,
} = stream[kState];
if (reader !== undefined && readableStreamHasBYOBReader(stream)) {
const readIntoRequests = reader[kState].readIntoRequests;
while (readIntoRequests.length)
readIntoRequests.shift()[kClose]();
}
return PromisePrototypeThen(
stream[kState].controller[kCancel](reason),
() => {});
}
function readableStreamClose(stream) {
assert(stream[kState].state === 'readable');
stream[kState].state = 'closed';
stream[kState].closedPromise?.resolve();
const {
reader,
} = stream[kState];
if (reader === undefined)
return;
reader[kState].close?.resolve();
if (readableStreamHasDefaultReader(stream)) {
const readRequests = reader[kState].readRequests;
while (readRequests.length)
readRequests.shift()[kClose]();
}
}
function readableStreamError(stream, error) {
assert(stream[kState].state === 'readable');
stream[kState].state = 'errored';
stream[kState].storedError = error;
const closedPromiseCache = stream[kState].closedPromise;
if (closedPromiseCache !== undefined) {
setPromiseHandled(closedPromiseCache.promise);
closedPromiseCache.reject(error);
}
const {
reader,
} = stream[kState];
if (reader === undefined)
return;
const closeCache = reader[kState].close;
if (closeCache !== undefined) {
setPromiseHandled(closeCache.promise);
closeCache.reject(error);
}
if (readableStreamHasDefaultReader(stream)) {
const readRequests = reader[kState].readRequests;
while (readRequests.length)
readRequests.shift()[kError](error);
} else {
assert(readableStreamHasBYOBReader(stream));
const readIntoRequests = reader[kState].readIntoRequests;
while (readIntoRequests.length)
readIntoRequests.shift()[kError](error);
}
}
function readableStreamHasDefaultReader(stream) {
const {
reader,
} = stream[kState];
if (reader === undefined)
return false;
return reader[kState] !== undefined &&
reader[kType] === 'ReadableStreamDefaultReader';
}
function readableStreamGetNumReadRequests(stream) {
assert(readableStreamHasDefaultReader(stream));
return stream[kState].reader[kState].readRequests.length;
}
function readableStreamHasBYOBReader(stream) {
const {
reader,
} = stream[kState];
if (reader === undefined)
return false;
return reader[kState] !== undefined &&
reader[kType] === 'ReadableStreamBYOBReader';
}
function readableStreamGetNumReadIntoRequests(stream) {
assert(readableStreamHasBYOBReader(stream));
return stream[kState].reader[kState].readIntoRequests.length;
}
function readableStreamFulfillReadRequest(stream, chunk, done) {
assert(readableStreamHasDefaultReader(stream));
const {
reader,
} = stream[kState];
assert(reader[kState].readRequests.length);
const readRequest = reader[kState].readRequests.shift();
// TODO(@jasnell): It's not clear under what exact conditions done
// will be true here. The spec requires this check but none of the
// WPT's or other tests trigger it. Will need to investigate how to
// get coverage for this.
if (done)
readRequest[kClose]();
else
readRequest[kChunk](chunk);
}
function readableStreamFulfillReadIntoRequest(stream, chunk, done) {
assert(readableStreamHasBYOBReader(stream));
const {
reader,
} = stream[kState];
assert(reader[kState].readIntoRequests.length);
const readIntoRequest = reader[kState].readIntoRequests.shift();
if (done)
readIntoRequest[kClose](chunk);
else
readIntoRequest[kChunk](chunk);
}
function readableStreamAddReadRequest(stream, readRequest) {
assert(readableStreamHasDefaultReader(stream));
assert(stream[kState].state === 'readable');
const readerState = stream[kState].reader[kState];
let readRequests = readerState.readRequests;
if (readRequests === kEmptyQueue)
readRequests = readerState.readRequests = new Queue();
readRequests.push(readRequest);
}
function readableStreamAddReadIntoRequest(stream, readIntoRequest) {
assert(readableStreamHasBYOBReader(stream));
assert(stream[kState].state !== 'errored');
const readerState = stream[kState].reader[kState];
let readIntoRequests = readerState.readIntoRequests;
if (readIntoRequests === kEmptyQueue)
readIntoRequests = readerState.readIntoRequests = new Queue();
readIntoRequests.push(readIntoRequest);
}
function readableStreamReaderGenericCancel(reader, reason) {
const {
stream,
} = reader[kState];
assert(stream !== undefined);
return readableStreamCancel(stream, reason);
}
function readableStreamReaderGenericInitialize(reader, stream) {
reader[kState].stream = stream;
stream[kState].reader = reader;
// The reader's closed promise (reader[kState].close) is materialized
// lazily by readerClosedPromise(); its settlement is fully derivable
// from the reader/stream state, so nothing is allocated here.
}
// Materializes the reader's [[closedPromise]] record on first
// observation. While unobserved, its settlement is derivable: a
// detached reader was released (rejected with the shared released
// error); otherwise it follows the stream state. The settle sites
// (readableStreamClose/Error, readableStreamReaderGenericRelease)
// only touch the record when it has been materialized.
function readerClosedPromise(reader) {
let close = reader[kState].close;
if (close === undefined) {
const stream = reader[kState].stream;
if (stream === undefined) {
close = {
promise: PromiseReject(lazyReadableReleasedError()),
resolve: undefined,
reject: undefined,
};
setPromiseHandled(close.promise);
} else {
switch (stream[kState].state) {
case 'readable':
close = PromiseWithResolvers();
break;
case 'closed':
close = {
promise: PromiseResolve(),
resolve: undefined,
reject: undefined,
};
break;
case 'errored':
close = {
promise: PromiseReject(stream[kState].storedError),
resolve: undefined,
reject: undefined,
};
setPromiseHandled(close.promise);
break;
}
}
reader[kState].close = close;
}
return close;
}
function readableStreamDefaultReaderRelease(reader) {
readableStreamReaderGenericRelease(reader);
readableStreamDefaultReaderErrorReadRequests(
reader,
lazyReadableReleasingError(),
);
}
function readableStreamDefaultReaderErrorReadRequests(reader, e) {
const readRequests = reader[kState].readRequests;
while (readRequests.length)
readRequests.shift()[kError](e);
}
function readableStreamBYOBReaderRelease(reader) {
readableStreamReaderGenericRelease(reader);
readableStreamBYOBReaderErrorReadIntoRequests(
reader,
lazyReadableReleasingError(),
);
}
function readableStreamBYOBReaderErrorReadIntoRequests(reader, e) {
const readIntoRequests = reader[kState].readIntoRequests;
while (readIntoRequests.length)
readIntoRequests.shift()[kError](e);
}
function readableStreamReaderGenericRelease(reader) {
const {
stream,
} = reader[kState];
assert(stream !== undefined);
assert(stream[kState].reader === reader);
const closeCache = reader[kState].close;
if (stream[kState].state === 'readable') {
if (closeCache !== undefined) {
closeCache.reject(lazyReadableReleasedError());
setPromiseHandled(closeCache.promise);
}
} else {
// The spec replaces [[closedPromise]] here. Dropping the cache makes
// the next observation derive the released rejection.
reader[kState].close = undefined;
}
stream[kState].controller[kRelease]();
stream[kState].reader = undefined;
reader[kState].stream = undefined;
}
// The argument validation half of ReadableStreamBYOBReader.prototype.read().
function readableStreamBYOBReaderReadView(reader, view, options) {
if (!isReadableStreamBYOBReader(reader))
throw new ERR_INVALID_THIS('ReadableStreamBYOBReader');
validateBuffer(view, 'view');
validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
const viewByteLength = ArrayBufferViewGetByteLength(view);
const viewBuffer = ArrayBufferViewGetBuffer(view);
if (isSharedArrayBuffer(viewBuffer)) {
throw new ERR_INVALID_ARG_VALUE(
'view',
view,
'must not be backed by a SharedArrayBuffer',
);
}
const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);
if (viewByteLength === 0 || viewBufferByteLength === 0) {
throw new ERR_INVALID_STATE.TypeError(
'View or Viewed ArrayBuffer is zero-length or detached');
}
// Supposed to assert here that the view's buffer is not
// detached, but there's no API available to use to check that.
const min = options?.min ?? 1;
validateNumber(min, 'options.min');
if (!NumberIsInteger(min))
throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer');
if (min <= 0)
throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0');
if (!isDataView(view)) {
if (min > TypedArrayPrototypeGetLength(view)) {
throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min);
}
} else if (min > viewByteLength) {
throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min);
}
if (reader[kState].stream === undefined) {
throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream');
}
const readIntoRequest = new ReadIntoRequest();
readableStreamBYOBReaderRead(reader, view, min, readIntoRequest);
return readIntoRequest.promise;
}
function readableStreamBYOBReaderRead(reader, view, min, readIntoRequest) {
const {
stream,
} = reader[kState];
assert(stream !== undefined);
stream[kState].disturbed = true;
if (stream[kState].state === 'errored') {
readIntoRequest[kError](stream[kState].storedError);
return;
}
readableByteStreamControllerPullInto(
stream[kState].controller,
view,
min,
readIntoRequest);
}
function readableStreamDefaultReaderRead(reader, readRequest) {
const {
stream,
} = reader[kState];
assert(stream !== undefined);
stream[kState].disturbed = true;
switch (stream[kState].state) {
case 'closed':
readRequest[kClose]();
break;
case 'errored':
readRequest[kError](stream[kState].storedError);
break;
case 'readable':
stream[kState].controller[kPull](readRequest);
}
}
function setupReadableStreamBYOBReader(reader, stream) {
if (isReadableStreamLocked(stream))
throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
const {
controller,
} = stream[kState];
if (!isReadableByteStreamController(controller))
throw new ERR_INVALID_ARG_VALUE('stream', stream, 'must be a byte stream');
readableStreamReaderGenericInitialize(reader, stream);
// The read-request queues use the same ring buffer as [[queue]], drained
// from a moving head rather than with ArrayPrototypeShift. Start from the
// shared immutable empty queue so acquiring a reader allocates no request
// storage until a read actually parks.
reader[kState].readIntoRequests = kEmptyQueue;
}
function setupReadableStreamDefaultReader(reader, stream) {
if (isReadableStreamLocked(stream))
throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
readableStreamReaderGenericInitialize(reader, stream);
reader[kState].readRequests = kEmptyQueue;
}
function readableStreamDefaultControllerClose(controller) {
if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller))
return;
controller[kState].closeRequested = true;
if (!controller[kState].queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(controller[kState].stream);
}
}
function readableStreamDefaultControllerEnqueue(controller, chunk) {
// Equivalent to readableStreamDefaultControllerCanCloseOrEnqueue()
// followed by isReadableStreamLocked() and
// readableStreamGetNumReadRequests(), but with the state loaded once:
// this runs for every enqueued chunk.
const controllerState = controller[kState];
const stream = controllerState.stream;
if (controllerState.closeRequested || stream[kState].state !== 'readable')
return;
const reader = stream[kState].reader;
if (reader !== undefined &&
reader[kState] !== undefined &&
reader[kType] === 'ReadableStreamDefaultReader' &&
reader[kState].readRequests.length) {
// Fulfilling a read request can run user code synchronously (the
// pipeTo read request invokes the sink's write algorithm), so the
// full ShouldCallPull predicate has to be re-evaluated afterwards.
readableStreamFulfillReadRequest(stream, chunk, false);
} else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) {
// The internal default size algorithm is never observable by user
// code, always returns 1, and cannot throw: enqueue with the
// constant size instead of calling it or validating the result.
// No user code runs between the guards at the top of this function
// and this point, so ShouldCallPull reduces to the started flag and
// the desired size (this branch implies no parked read requests,
// ruling out the reader arm of the predicate).
materializeQueue(controllerState).pushPair(chunk, 1);
controllerState.queueTotalSize++;
if (controllerState.started &&
controllerState.highWaterMark - controllerState.queueTotalSize > 0) {
readableStreamDefaultControllerPull(controller);
}
return;
} else {
// The user-supplied size algorithm may run arbitrary code, so the
// full ShouldCallPull predicate has to be re-evaluated afterwards.
try {
const chunkSize =
FunctionPrototypeCall(
controllerState.sizeAlgorithm,
undefined,
chunk);
enqueueValueWithSize(controller, chunk, chunkSize);
} catch (error) {
readableStreamDefaultControllerError(controller, error);
throw error;
}
}
readableStreamDefaultControllerCallPullIfNeeded(controller);
}
function readableStreamDefaultControllerHasBackpressure(controller) {
return !readableStreamDefaultControllerShouldCallPull(controller);
}
function readableStreamDefaultControllerCanCloseOrEnqueue(controller) {
const {
stream,
} = controller[kState];
return !controller[kState].closeRequested &&
stream[kState].state === 'readable';
}
function readableStreamDefaultControllerGetDesiredSize(controller) {
const {
stream,
highWaterMark,
queueTotalSize,
} = controller[kState];
switch (stream[kState].state) {
case 'errored': return null;
case 'closed': return 0;
default:
return highWaterMark - queueTotalSize;
}
}
function readableStreamDefaultControllerShouldCallPull(controller) {
// Single-pass version of the spec's predicate chain (CanCloseOrEnqueue,
// IsLocked, HasDefaultReader, GetNumReadRequests, GetDesiredSize): this
// runs at least once per chunk on every default-stream path. The
// desired-size computation is inlined because the stream state is
// already known to be 'readable' here.
const controllerState = controller[kState];
const stream = controllerState.stream;
if (controllerState.closeRequested ||
stream[kState].state !== 'readable' ||
!controllerState.started)
return false;
const reader = stream[kState].reader;
if (reader !== undefined &&
reader[kState] !== undefined &&
reader[kType] === 'ReadableStreamDefaultReader' &&
reader[kState].readRequests.length) {
return true;
}
return controllerState.highWaterMark - controllerState.queueTotalSize > 0;
}
function readableStreamDefaultControllerCallPullIfNeeded(controller) {
if (!readableStreamDefaultControllerShouldCallPull(controller))
return;
readableStreamDefaultControllerPull(controller);
}
// The ShouldCallPull half of CallPullIfNeeded, split out so that callers
// that have already established the predicate from state in scope (the
// enqueue path above) can skip re-running it.
function readableStreamDefaultControllerPull(controller) {
// A source without pull() has nothing to run: skip the pulling/pullAgain
// bookkeeping, the reaction closures and the microtask that would only
// clear the flag again. Push-style sources hit this on every read.
if (controller[kState].pullAlgorithm === nonOpCallback)
return;
if (controller[kState].pulling) {
controller[kState].pullAgain = true;
return;
}
assert(!controller[kState].pullAgain);
controller[kState].pulling = true;
if (controller[kState].pullFulfilled === undefined) {
// The pull reaction closures only capture the controller, so they are
// created once on the first pull and reused for every subsequent pull
// instead of allocating two fresh closures per chunk.
controller[kState].pullFulfilled = () => {
controller[kState].pulling = false;
if (controller[kState].pullAgain) {
controller[kState].pullAgain = false;
readableStreamDefaultControllerCallPullIfNeeded(controller);
}
};
controller[kState].pullRejected =
(error) => readableStreamDefaultControllerError(controller, error);
}
// The pull algorithm may be a raw callback (a wrapped user source.pull
// returns its result uncoerced; a synchronous throw surfaces here) or an
// internal algorithm that always returns a promise; thenAlgorithmResult
// handles both.
let result;
try {
result = controller[kState].pullAlgorithm(controller);
} catch (error) {
result = PromiseReject(error);
}
thenAlgorithmResult(
result,
controller[kState].pullFulfilled,
controller[kState].pullRejected);
}
function readableStreamDefaultControllerClearAlgorithms(controller) {
controller[kState].pullAlgorithm = undefined;
controller[kState].cancelAlgorithm = undefined;
controller[kState].sizeAlgorithm = undefined;
}
function readableStreamDefaultControllerError(controller, error) {
const {
stream,
} = controller[kState];
if (stream[kState].state === 'readable') {
resetQueue(controller);
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamError(stream, error);
}
}
function readableStreamDefaultControllerCancelSteps(controller, reason) {
resetQueue(controller);
const result = controller[kState].cancelAlgorithm(reason, controller);
readableStreamDefaultControllerClearAlgorithms(controller);
return result;
}
function readableStreamDefaultControllerPullSteps(controller, readRequest) {
const {
stream,
queue,
} = controller[kState];
if (queue.length) {
const chunk = dequeueValue(controller);
if (controller[kState].closeRequested && !queue.length) {
readableStreamDefaultControllerClearAlgorithms(controller);
readableStreamClose(stream);
} else if (!controller[kState].closeRequested &&
controller[kState].started &&
controller[kState].highWaterMark -
controller[kState].queueTotalSize > 0) {
// Reduced ShouldCallPull: the state is known to be readable and the
// queue was non-empty, so no read requests can be parked.
readableStreamDefaultControllerPull(controller);
}
readRequest[kChunk](chunk);
return;
}
readableStreamAddReadRequest(stream, readRequest);
// Reduced ShouldCallPull: the state is known to be readable, an empty
// queue in that state implies close has not been requested (close with
// an empty queue closes the stream immediately), and the read request
// parked above already satisfies the reader-with-pending-reads arm.
if (controller[kState].started)
readableStreamDefaultControllerPull(controller);
}
function setupReadableStreamDefaultController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
sizeAlgorithm) {
assert(stream[kState].controller === undefined);
controller[kState] = {
cancelAlgorithm,
closeRequested: false,
highWaterMark,
pullAgain: false,
pullAlgorithm,
pulling: false,
pullFulfilled: undefined,
pullRejected: undefined,
queue: kEmptyQueue,
queueTotalSize: 0,
started: false,
sizeAlgorithm,
stream,
// The TransformStream this controller is the source of, if any; its
// shared source algorithms reach the stream through it.
transformStream: undefined,
};
stream[kState].controller = controller;
const startResult = startAlgorithm();
if (startResult === kDeferredStart)
return;
// A non-thenable start result guarantees fulfillment, and no .then
// lookup on it is observable.
const startFulfilled = startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function');
// The started flag only gates calls into the pull algorithm, so for a
// source without pull() the post-start step has nothing observable
// left to do: the flag is set right away instead of from a microtask.
if (startFulfilled && pullAlgorithm === nonOpCallback) {
controller[kState].started = true;
return;
}
const started = () => readableStreamDefaultControllerStarted(controller);
if (startFulfilled) {
// The post-start step runs at the exact microtask position the
// promise reaction would have had.
queueMicrotask(started);
return;
}
// The wrapper promise matches the reference implementation's
// promiseResolvedWith(), whose extra microtask hops WPT relies on.
PromisePrototypeThen(
new Promise((r) => r(startResult)),
started,
(error) => readableStreamDefaultControllerError(controller, error));
}
// The post-start steps; transform streams deliver them directly (see
// transformStreamStart).
function readableStreamDefaultControllerStarted(controller) {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableStreamDefaultControllerCallPullIfNeeded(controller);
}
function setupReadableStreamDefaultControllerFromSource(
stream,
source,
highWaterMark,
sizeAlgorithm) {
const controller = new ReadableStreamDefaultController(kSkipThrow);
const start = source?.start;
const pull = source?.pull;
const cancel = source?.cancel;
const startAlgorithm = start ?
FunctionPrototypeBind(start, source, controller) :
nonOpCallback;
const pullAlgorithm = pull ?
createRawCallback1Param('source.pull', pull, source) :
nonOpCallback;
const cancelAlgorithm = cancel ?
createPromiseCallback1Param('source.cancel', cancel, source) :
nonOpCancel;
setupReadableStreamDefaultController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
sizeAlgorithm);
}
function readableByteStreamControllerClose(controller) {
const {
closeRequested,
pendingPullIntos,
queueTotalSize,
stream,
} = controller[kState];
if (closeRequested || stream[kState].state !== 'readable')
return;
if (queueTotalSize) {
controller[kState].closeRequested = true;
return;
}
if (pendingPullIntos.length) {
const firstPendingPullInto = pendingPullIntos.peek();
if (firstPendingPullInto.bytesFilled % firstPendingPullInto.elementSize !== 0) {
const error = new ERR_INVALID_STATE.TypeError('Partial read');
readableByteStreamControllerError(controller, error);
throw error;
}
}
readableByteStreamControllerClearAlgorithms(controller);
readableStreamClose(stream);
}
function readableByteStreamControllerCommitPullIntoDescriptor(stream, desc) {
assert(stream[kState].state !== 'errored');
assert(desc.type !== 'none');
let done = false;
if (stream[kState].state === 'closed') {
assert(desc.bytesFilled % desc.elementSize === 0);
done = true;
}
const filledView =
readableByteStreamControllerConvertPullIntoDescriptor(desc);
if (desc.type === 'default') {
readableStreamFulfillReadRequest(stream, filledView, done);
} else {
assert(desc.type === 'byob');
readableStreamFulfillReadIntoRequest(stream, filledView, done);
}
}
function readableByteStreamControllerCommitPullIntoDescriptors(stream, descriptors) {
for (let i = 0; i < descriptors.length; ++i) {
readableByteStreamControllerCommitPullIntoDescriptor(
stream,
descriptors[i],
);
}
}
function readableByteStreamControllerInvalidateBYOBRequest(controller) {
if (controller[kState].byobRequest === null)
return;
controller[kState].byobRequest[kState].controller = undefined;
controller[kState].byobRequest[kState].view = null;
controller[kState].byobRequest = null;
}
function readableByteStreamControllerClearAlgorithms(controller) {
controller[kState].pullAlgorithm = undefined;
controller[kState].cancelAlgorithm = undefined;
}
function readableByteStreamControllerClearPendingPullIntos(controller) {
readableByteStreamControllerInvalidateBYOBRequest(controller);
controller[kState].pendingPullIntos = kEmptyQueue;
}
function readableByteStreamControllerGetDesiredSize(controller) {
const {
stream,
highWaterMark,
queueTotalSize,
} = controller[kState];
switch (stream[kState].state) {
case 'errored': return null;
case 'closed': return 0;
default: return highWaterMark - queueTotalSize;
}
}
function readableByteStreamControllerShouldCallPull(controller) {
// Single-pass version of the spec's predicate chain (HasDefaultReader,
// GetNumReadRequests, HasBYOBReader, GetNumReadIntoRequests,
// GetDesiredSize): this runs at least once per chunk on every byte
// stream path. The desired-size computation is inlined because the
// stream state is already known to be 'readable' here.
const controllerState = controller[kState];
const stream = controllerState.stream;
if (stream[kState].state !== 'readable' ||
controllerState.closeRequested ||
!controllerState.started) {
return false;
}
const reader = stream[kState].reader;
if (reader !== undefined && reader[kState] !== undefined) {
const type = reader[kType];
if (type === 'ReadableStreamDefaultReader') {
if (reader[kState].readRequests.length) return true;
} else if (type === 'ReadableStreamBYOBReader') {
if (reader[kState].readIntoRequests.length) return true;
}
}
return controllerState.highWaterMark - controllerState.queueTotalSize > 0;
}
function readableByteStreamControllerHandleQueueDrain(controller) {
const {
closeRequested,
queueTotalSize,
stream,
} = controller[kState];
assert(stream[kState].state === 'readable');
if (!queueTotalSize && closeRequested) {
readableByteStreamControllerClearAlgorithms(controller);
readableStreamClose(stream);
return;
}
readableByteStreamControllerCallPullIfNeeded(controller);
}
function readableByteStreamControllerPullInto(
controller,
view,
min,
readIntoRequest) {
const {
closeRequested,
stream,
pendingPullIntos,
} = controller[kState];
let elementSize = 1;
let ctor = DataView;
if (isArrayBufferView(view) && !isDataView(view)) {
elementSize = view.constructor.BYTES_PER_ELEMENT;
ctor = view.constructor;
}
const minimumFill = min * elementSize;
assert(minimumFill >= elementSize && minimumFill <= view.byteLength);
assert(minimumFill % elementSize === 0);
const buffer = ArrayBufferViewGetBuffer(view);
const byteOffset = ArrayBufferViewGetByteOffset(view);
const byteLength = ArrayBufferViewGetByteLength(view);
const bufferByteLength = ArrayBufferPrototypeGetByteLength(buffer);
let transferredBuffer;
try {
transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
} catch (error) {
readIntoRequest[kError](error);
return;
}
const desc = {
buffer: transferredBuffer,
bufferByteLength,
byteOffset,
byteLength,
bytesFilled: 0,
minimumFill,
elementSize,
ctor,
type: 'byob',
};
if (pendingPullIntos.length) {
pendingPullIntos.push(desc);
readableStreamAddReadIntoRequest(stream, readIntoRequest);
return;
}
if (stream[kState].state === 'closed') {
const emptyView = new ctor(desc.buffer, byteOffset, 0);
readIntoRequest[kClose](emptyView);
return;
}
if (controller[kState].queueTotalSize) {
if (readableByteStreamControllerFillPullIntoDescriptorFromQueue(
controller,
desc)) {
const filledView =
readableByteStreamControllerConvertPullIntoDescriptor(desc);
readableByteStreamControllerHandleQueueDrain(controller);
readIntoRequest[kChunk](filledView);
return;
}
if (closeRequested) {
const error = new ERR_INVALID_STATE.TypeError('ReadableStream closed');
readableByteStreamControllerError(controller, error);
readIntoRequest[kError](error);
return;
}
}
materializePendingPullIntos(controller[kState]).push(desc);
readableStreamAddReadIntoRequest(stream, readIntoRequest);
readableByteStreamControllerCallPullIfNeeded(controller);
}
// Pending pull-into descriptor queues start out as (and are reset to)
// the shared immutable empty queue so that constructing a byte stream
// never allocates descriptor storage; the two push sites that can
// observe an empty queue materialize a real Queue on first use.
function materializePendingPullIntos(state) {
const pendingPullIntos = state.pendingPullIntos;
if (pendingPullIntos === kEmptyQueue)
return state.pendingPullIntos = new Queue();
return pendingPullIntos;
}
function readableByteStreamControllerRespondInternal(controller, bytesWritten) {
const {
stream,
pendingPullIntos,
} = controller[kState];
const desc = pendingPullIntos.peek();
readableByteStreamControllerInvalidateBYOBRequest(controller);
if (stream[kState].state === 'closed') {
if (bytesWritten)
throw new ERR_INVALID_STATE.TypeError(
'Controller is closed but view is not zero-length');
readableByteStreamControllerRespondInClosedState(controller, desc);
} else {
assert(stream[kState].state === 'readable');
if (!bytesWritten)
throw new ERR_INVALID_STATE.TypeError('View cannot be zero-length');
readableByteStreamControllerRespondInReadableState(
controller,
bytesWritten,
desc);
}
readableByteStreamControllerCallPullIfNeeded(controller);
}
function readableByteStreamControllerRespond(controller, bytesWritten) {
const {
pendingPullIntos,
stream,
} = controller[kState];
assert(pendingPullIntos.length);
const desc = pendingPullIntos.peek();
if (stream[kState].state === 'closed') {
if (bytesWritten !== 0)
throw new ERR_INVALID_ARG_VALUE('bytesWritten', bytesWritten);
} else {
assert(stream[kState].state === 'readable');
if (!bytesWritten)
throw new ERR_INVALID_ARG_VALUE('bytesWritten', bytesWritten);
if ((desc.bytesFilled + bytesWritten) > desc.byteLength)
throw new ERR_INVALID_ARG_VALUE.RangeError('bytesWritten', bytesWritten);
}
desc.buffer = ArrayBufferPrototypeTransfer(desc.buffer);
readableByteStreamControllerRespondInternal(controller, bytesWritten);
}
function readableByteStreamControllerRespondInClosedState(controller, desc) {
assert(desc.bytesFilled % desc.elementSize === 0);
if (desc.type === 'none') {
readableByteStreamControllerShiftPendingPullInto(controller);
}
const {
stream,
} = controller[kState];
if (readableStreamHasBYOBReader(stream)) {
const filledPullIntos = [];
for (let i = 0; i < readableStreamGetNumReadIntoRequests(stream); ++i) {
ArrayPrototypePush(filledPullIntos, readableByteStreamControllerShiftPendingPullInto(controller));
}
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
}
}
function readableByteStreamControllerFillHeadPullIntoDescriptor(
controller,
size,
desc) {
const {
pendingPullIntos,
byobRequest,
} = controller[kState];
assert(!pendingPullIntos.length || pendingPullIntos.peek() === desc);
assert(byobRequest === null);
desc.bytesFilled += size;
}
function readableByteStreamControllerEnqueue(controller, chunk) {
const {
closeRequested,
pendingPullIntos,
queue,
stream,
} = controller[kState];
const buffer = ArrayBufferViewGetBuffer(chunk);
const byteOffset = ArrayBufferViewGetByteOffset(chunk);
const byteLength = ArrayBufferViewGetByteLength(chunk);
if (closeRequested || stream[kState].state !== 'readable')
return;
const transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
if (pendingPullIntos.length) {
const firstPendingPullInto = pendingPullIntos.peek();
if (ArrayBufferPrototypeGetDetached(firstPendingPullInto.buffer)) {
throw new ERR_INVALID_STATE.TypeError(
'Destination ArrayBuffer is detached',
);
}
readableByteStreamControllerInvalidateBYOBRequest(controller);
firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer(
firstPendingPullInto.buffer,
);
if (firstPendingPullInto.type === 'none') {
readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
controller,
firstPendingPullInto,
);
}
}
// Single consolidated pass over the reader state. The spec routes this
// through HasDefaultReader / ProcessReadRequestsUsingQueue /
// GetNumReadRequests / FulfillReadRequest, which would re-run the same
// reader brand check and re-load the read request list four times on
// this per-chunk path.
const { reader } = stream[kState];
if (reader !== undefined &&
reader[kState] !== undefined &&
reader[kType] === 'ReadableStreamDefaultReader') {
const { readRequests } = reader[kState];
if (readRequests.length && controller[kState].queueTotalSize > 0) {
// Only possible when data was enqueued while the stream was not
// being read; read requests otherwise never coexist with a
// non-empty queue.
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
}
if (!readRequests.length) {
readableByteStreamControllerEnqueueChunkToQueue(
controller,
transferredBuffer,
byteOffset,
byteLength);
} else {
assert(!queue.length);
if (pendingPullIntos.length) {
assert(pendingPullIntos.peek().type === 'default');
readableByteStreamControllerShiftPendingPullInto(controller);
}
const transferredView =
new Uint8Array(transferredBuffer, byteOffset, byteLength);
const readRequest = readRequests.shift();
readRequest[kChunk](transferredView);
}
} else if (readableStreamHasBYOBReader(stream)) {
readableByteStreamControllerEnqueueChunkToQueue(
controller,
transferredBuffer,
byteOffset,
byteLength);
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
controller);
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
} else {
assert(!isReadableStreamLocked(stream));
readableByteStreamControllerEnqueueChunkToQueue(
controller,
transferredBuffer,
byteOffset,
byteLength);
}
readableByteStreamControllerCallPullIfNeeded(controller);
}
function readableByteStreamControllerEnqueueClonedChunkToQueue(
controller,
buffer,
byteOffset,
byteLength,
) {
let cloneResult;
try {
cloneResult = ArrayBufferPrototypeSlice(
buffer,
byteOffset,
byteOffset + byteLength,
);
} catch (error) {
readableByteStreamControllerError(controller, error);
throw error;
}
readableByteStreamControllerEnqueueChunkToQueue(
controller,
cloneResult,
0,
byteLength,
);
}
function readableByteStreamControllerEnqueueChunkToQueue(
controller,
buffer,
byteOffset,
byteLength) {
const state = controller[kState];
materializeQueue(state).push({
buffer,
byteOffset,
byteLength,
});
state.queueTotalSize += byteLength;
}
function readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
controller,
desc,
) {
const {
buffer,
byteOffset,
bytesFilled,
type,
} = desc;
assert(type === 'none');
if (bytesFilled > 0) {
readableByteStreamControllerEnqueueClonedChunkToQueue(
controller,
buffer,
byteOffset,
bytesFilled,
);
}
readableByteStreamControllerShiftPendingPullInto(controller);
}
function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
controller,
desc) {
const {
buffer,
byteLength,
byteOffset,
bytesFilled,
minimumFill,
elementSize,
} = desc;
const maxBytesToCopy = MathMin(
controller[kState].queueTotalSize,
byteLength - bytesFilled);
const maxBytesFilled = bytesFilled + maxBytesToCopy;
const maxAlignedBytes = maxBytesFilled - (maxBytesFilled % elementSize);
let totalBytesToCopyRemaining = maxBytesToCopy;
let ready = false;
assert(!ArrayBufferPrototypeGetDetached(buffer));
assert(bytesFilled < minimumFill);
if (maxAlignedBytes >= minimumFill) {
totalBytesToCopyRemaining = maxAlignedBytes - bytesFilled;
ready = true;
}
const {
queue,
} = controller[kState];
while (totalBytesToCopyRemaining) {
const headOfQueue = queue.peek();
const bytesToCopy = MathMin(
totalBytesToCopyRemaining,
headOfQueue.byteLength);
const destStart = byteOffset + desc.bytesFilled;
assert(canCopyArrayBuffer(
buffer,
destStart,
headOfQueue.buffer,
headOfQueue.byteOffset,
bytesToCopy));
copyArrayBuffer(
buffer,
destStart,
headOfQueue.buffer,
headOfQueue.byteOffset,
bytesToCopy);
if (headOfQueue.byteLength === bytesToCopy) {
queue.shift();
} else {
headOfQueue.byteOffset += bytesToCopy;
headOfQueue.byteLength -= bytesToCopy;
}
controller[kState].queueTotalSize -= bytesToCopy;
readableByteStreamControllerFillHeadPullIntoDescriptor(
controller,
bytesToCopy,
desc);
totalBytesToCopyRemaining -= bytesToCopy;
}
if (!ready) {
assert(!controller[kState].queueTotalSize);
assert(desc.bytesFilled > 0);
assert(desc.bytesFilled < minimumFill);
}
return ready;
}
function readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
controller) {
const {
closeRequested,
pendingPullIntos,
} = controller[kState];
assert(!closeRequested);
const filledPullIntos = [];
while (pendingPullIntos.length) {
if (!controller[kState].queueTotalSize)
break;
const desc = pendingPullIntos.peek();
if (readableByteStreamControllerFillPullIntoDescriptorFromQueue(
controller,
desc)) {
readableByteStreamControllerShiftPendingPullInto(controller);
ArrayPrototypePush(filledPullIntos, desc);
}
}
return filledPullIntos;
}
function readableByteStreamControllerRespondInReadableState(
controller,
bytesWritten,
desc) {
const {
stream,
} = controller[kState];
const {
buffer,
bytesFilled,
byteLength,
type,
} = desc;
if (bytesFilled + bytesWritten > byteLength)
throw new ERR_INVALID_STATE.RangeError('The buffer size is invalid');
readableByteStreamControllerFillHeadPullIntoDescriptor(
controller,
bytesWritten,
desc);
if (type === 'none') {
readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
controller,
desc,
);
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
controller,
);
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
return;
}
if (desc.bytesFilled < desc.minimumFill)
return;
readableByteStreamControllerShiftPendingPullInto(controller);
const remainderSize = desc.bytesFilled % desc.elementSize;
if (remainderSize) {
const end = desc.byteOffset + desc.bytesFilled;
const start = end - remainderSize;
const remainder =
ArrayBufferPrototypeSlice(
buffer,
start,
end);
readableByteStreamControllerEnqueueChunkToQueue(
controller,
remainder,
0,
ArrayBufferPrototypeGetByteLength(remainder));
}
desc.bytesFilled -= remainderSize;
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
readableByteStreamControllerCommitPullIntoDescriptor(stream, desc);
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
}
function readableByteStreamControllerRespondWithNewView(controller, view) {
const {
stream,
pendingPullIntos,
} = controller[kState];
assert(pendingPullIntos.length);
const desc = pendingPullIntos.peek();
assert(stream[kState].state !== 'errored');
const viewByteLength = ArrayBufferViewGetByteLength(view);
const viewByteOffset = ArrayBufferViewGetByteOffset(view);
const viewBuffer = ArrayBufferViewGetBuffer(view);
const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);
if (stream[kState].state === 'closed') {
if (viewByteLength !== 0)
throw new ERR_INVALID_STATE.TypeError('View is not zero-length');
} else {
assert(stream[kState].state === 'readable');
if (viewByteLength === 0)
throw new ERR_INVALID_STATE.TypeError('View is zero-length');
}
const {
byteOffset,
byteLength,
bytesFilled,
bufferByteLength,
} = desc;
if (byteOffset + bytesFilled !== viewByteOffset)
throw new ERR_INVALID_ARG_VALUE.RangeError('view', view);
if (bytesFilled + viewByteLength > byteLength)
throw new ERR_INVALID_ARG_VALUE.RangeError('view', view);
if (bufferByteLength !== viewBufferByteLength)
throw new ERR_INVALID_ARG_VALUE.RangeError('view', view);
desc.buffer = ArrayBufferPrototypeTransfer(viewBuffer);
readableByteStreamControllerRespondInternal(controller, viewByteLength);
}
function readableByteStreamControllerShiftPendingPullInto(controller) {
assert(controller[kState].byobRequest === null);
return controller[kState].pendingPullIntos.shift();
}
function readableByteStreamControllerCallPullIfNeeded(controller) {
if (!readableByteStreamControllerShouldCallPull(controller))
return;
// See readableStreamDefaultControllerPull.
if (controller[kState].pullAlgorithm === nonOpCallback)
return;
if (controller[kState].pulling) {
controller[kState].pullAgain = true;
return;
}
assert(!controller[kState].pullAgain);
controller[kState].pulling = true;
if (controller[kState].pullFulfilled === undefined) {
// See readableStreamDefaultControllerCallPullIfNeeded: created once,
// reused for every pull.
controller[kState].pullFulfilled = () => {
controller[kState].pulling = false;
if (controller[kState].pullAgain) {
controller[kState].pullAgain = false;
readableByteStreamControllerCallPullIfNeeded(controller);
}
};
controller[kState].pullRejected =
(error) => readableByteStreamControllerError(controller, error);
}
// See readableStreamDefaultControllerPull for the raw-callback contract.
let result;
try {
result = controller[kState].pullAlgorithm(controller);
} catch (error) {
result = PromiseReject(error);
}
thenAlgorithmResult(
result,
controller[kState].pullFulfilled,
controller[kState].pullRejected);
}
function readableByteStreamControllerError(controller, error) {
const {
stream,
} = controller[kState];
if (stream[kState].state !== 'readable')
return;
readableByteStreamControllerClearPendingPullIntos(controller);
resetQueue(controller);
readableByteStreamControllerClearAlgorithms(controller);
readableStreamError(stream, error);
}
function readableByteStreamControllerCancelSteps(controller, reason) {
readableByteStreamControllerClearPendingPullIntos(controller);
resetQueue(controller);
const result = controller[kState].cancelAlgorithm(reason);
readableByteStreamControllerClearAlgorithms(controller);
return result;
}
// Dequeues the first chunk of the byte queue as a Uint8Array view,
// handling queue drain (close-on-empty or pull) before the view is
// created. This is the [[queueTotalSize]] > 0 arm of the byte
// controller's pull steps; it is also called directly from the
// buffered fast paths in ReadableStreamDefaultReader.read() and the
// async iterator, which resolve with the view without allocating a
// read request.
function readableByteStreamControllerDequeueChunk(controller) {
assert(controller[kState].queueTotalSize > 0);
const {
buffer,
byteOffset,
byteLength,
} = controller[kState].queue.shift();
controller[kState].queueTotalSize -= byteLength;
readableByteStreamControllerHandleQueueDrain(controller);
return new Uint8Array(buffer, byteOffset, byteLength);
}
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
readRequest[kChunk](readableByteStreamControllerDequeueChunk(controller));
}
function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
const {
stream,
queueTotalSize,
} = controller[kState];
const { reader } = stream[kState];
assert(isReadableStreamDefaultReader(reader));
while (reader[kState].readRequests.length > 0) {
if (queueTotalSize === 0) {
return;
}
readableByteStreamControllerFillReadRequestFromQueue(
controller,
reader[kState].readRequests.shift(),
);
}
}
function readableByteStreamControllerPullSteps(controller, readRequest) {
const {
queueTotalSize,
stream,
} = controller[kState];
assert(readableStreamHasDefaultReader(stream));
if (queueTotalSize) {
assert(!readableStreamGetNumReadRequests(stream));
readableByteStreamControllerFillReadRequestFromQueue(
controller,
readRequest,
);
return;
}
const {
autoAllocateChunkSize,
} = controller[kState];
if (autoAllocateChunkSize !== undefined) {
try {
const buffer = new ArrayBuffer(autoAllocateChunkSize);
materializePendingPullIntos(controller[kState]).push(
{
buffer,
bufferByteLength: autoAllocateChunkSize,
byteOffset: 0,
byteLength: autoAllocateChunkSize,
bytesFilled: 0,
minimumFill: 1,
elementSize: 1,
ctor: Uint8Array,
type: 'default',
});
} catch (error) {
readRequest[kError](error);
return;
}
}
readableStreamAddReadRequest(stream, readRequest);
readableByteStreamControllerCallPullIfNeeded(controller);
}
function setupReadableByteStreamController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
autoAllocateChunkSize) {
assert(stream[kState].controller === undefined);
if (autoAllocateChunkSize !== undefined) {
assert(NumberIsInteger(autoAllocateChunkSize));
assert(autoAllocateChunkSize > 0);
}
controller[kState] = {
byobRequest: null,
closeRequested: false,
pullAgain: false,
pulling: false,
pullFulfilled: undefined,
pullRejected: undefined,
started: false,
stream,
queue: kEmptyQueue,
queueTotalSize: 0,
highWaterMark,
pullAlgorithm,
cancelAlgorithm,
autoAllocateChunkSize,
pendingPullIntos: kEmptyQueue,
};
stream[kState].controller = controller;
const startResult = startAlgorithm();
// See setupReadableStreamDefaultController.
const startFulfilled = startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function');
if (startFulfilled && pullAlgorithm === nonOpCallback) {
controller[kState].started = true;
return;
}
const started = () => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableByteStreamControllerCallPullIfNeeded(controller);
};
if (startFulfilled) {
queueMicrotask(started);
return;
}
PromisePrototypeThen(
new Promise((r) => r(startResult)),
started,
(error) => readableByteStreamControllerError(controller, error));
}
function setupReadableByteStreamControllerFromSource(
stream,
source,
highWaterMark) {
const controller = new ReadableByteStreamController(kSkipThrow);
const start = source?.start;
const pull = source?.pull;
const cancel = source?.cancel;
const autoAllocateChunkSize = source?.autoAllocateChunkSize;
const startAlgorithm = start ?
FunctionPrototypeBind(start, source, controller) :
nonOpCallback;
const pullAlgorithm = pull ?
createRawCallback1Param('source.pull', pull, source) :
nonOpCallback;
const cancelAlgorithm = cancel ?
createPromiseCallback1Param('source.cancel', cancel, source) :
nonOpCancel;
if (autoAllocateChunkSize === 0) {
throw new ERR_INVALID_ARG_VALUE(
'source.autoAllocateChunkSize',
autoAllocateChunkSize);
}
setupReadableByteStreamController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
autoAllocateChunkSize);
}
module.exports = {
ReadableStream,
ReadableStreamDefaultReader,
ReadableStreamBYOBReader,
ReadableStreamBYOBRequest,
ReadableByteStreamController,
ReadableStreamDefaultController,
TransferredReadableStream,
// Exported Brand Checks
isReadableStream,
isReadableByteStreamController,
isReadableStreamBYOBRequest,
isReadableStreamDefaultReader,
isReadableStreamBYOBReader,
isWritableStreamDefaultWriter,
isWritableStreamDefaultController,
readableStreamPipeTo,
readableStreamTee,
readableByteStreamControllerConvertPullIntoDescriptor,
isReadableStreamLocked,
readableStreamCancel,
readableStreamClose,
readableStreamError,
readableStreamHasDefaultReader,
readableStreamGetNumReadRequests,
readableStreamHasBYOBReader,
readableStreamGetNumReadIntoRequests,
readableStreamFulfillReadRequest,
readableStreamFulfillReadIntoRequest,
readableStreamAddReadRequest,
readableStreamAddReadIntoRequest,
readableStreamReaderGenericCancel,
readableStreamReaderGenericInitialize,
readableStreamReaderGenericRelease,
readableStreamBYOBReaderRead,
readableStreamDefaultReaderRead,
kChunk,
kClose,
kError,
setupReadableStreamBYOBReader,
setupReadableStreamDefaultReader,
readableStreamDefaultControllerClose,
readableStreamDefaultControllerEnqueue,
readableStreamDefaultControllerHasBackpressure,
readableStreamDefaultControllerCanCloseOrEnqueue,
readableStreamDefaultControllerGetDesiredSize,
readableStreamDefaultControllerShouldCallPull,
readableStreamDefaultControllerCallPullIfNeeded,
readableStreamDefaultControllerClearAlgorithms,
readableStreamDefaultControllerError,
readableStreamDefaultControllerCancelSteps,
readableStreamDefaultControllerPullSteps,
readableStreamDefaultControllerStarted,
setupReadableStreamDefaultController,
setupReadableStreamDefaultControllerFromSource,
readableByteStreamControllerClose,
readableByteStreamControllerCommitPullIntoDescriptor,
readableByteStreamControllerInvalidateBYOBRequest,
readableByteStreamControllerClearAlgorithms,
readableByteStreamControllerClearPendingPullIntos,
readableByteStreamControllerGetDesiredSize,
readableByteStreamControllerShouldCallPull,
readableByteStreamControllerHandleQueueDrain,
readableByteStreamControllerPullInto,
readableByteStreamControllerRespondInternal,
readableByteStreamControllerRespond,
readableByteStreamControllerRespondInClosedState,
readableByteStreamControllerFillHeadPullIntoDescriptor,
readableByteStreamControllerEnqueue,
readableByteStreamControllerEnqueueChunkToQueue,
readableByteStreamControllerFillPullIntoDescriptorFromQueue,
readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue,
readableByteStreamControllerRespondInReadableState,
readableByteStreamControllerRespondWithNewView,
readableByteStreamControllerShiftPendingPullInto,
readableByteStreamControllerCallPullIfNeeded,
readableByteStreamControllerError,
readableByteStreamControllerCancelSteps,
readableByteStreamControllerPullSteps,
setupReadableByteStreamController,
setupReadableByteStreamControllerFromSource,
createReadableStream,
createReadableByteStream,
};