'use strict';
const {
FunctionPrototypeCall,
ObjectDefineProperties,
ObjectSetPrototypeOf,
Promise,
PromisePrototypeThen,
PromiseReject,
PromiseResolve,
PromiseWithResolvers,
Symbol,
SymbolToStringTag,
} = primordials;
const {
codes: {
ERR_ILLEGAL_CONSTRUCTOR,
ERR_INVALID_ARG_VALUE,
ERR_INVALID_STATE,
ERR_INVALID_THIS,
},
} = require('internal/errors');
const {
DOMException,
} = internalBinding('messaging');
const AsyncContextFrame = require('internal/async_context_frame');
const {
customInspectSymbol: kInspect,
kEmptyObject,
kEnumerableProperty,
} = require('internal/util');
const {
validateObject,
kValidateObjectAllowObjects,
kValidateObjectAllowObjectsAndNull,
} = require('internal/validators');
const {
kDeserialize,
kTransfer,
kTransferList,
markTransferMode,
} = require('internal/worker/js_transferable');
const {
createPromiseCallback1Param,
createRawCallback2Params,
customInspect,
deferredStartAlgorithm,
extractHighWaterMark,
extractSizeAlgorithm,
getNonWritablePropertyDescriptor,
isBrandCheck,
kParkedAlgorithmResult,
kResolvedPromise,
kState,
kType,
nonOpCancel,
nonOpFlush,
} = require('internal/webstreams/util');
const {
createReadableStream,
readableStreamDefaultControllerCanCloseOrEnqueue,
readableStreamDefaultControllerClose,
readableStreamDefaultControllerEnqueue,
readableStreamDefaultControllerError,
readableStreamDefaultControllerGetDesiredSize,
readableStreamDefaultControllerHasBackpressure,
readableStreamDefaultControllerStarted,
} = require('internal/webstreams/readablestream');
const {
createWritableStream,
writableStreamDefaultControllerErrorIfNeeded,
writableStreamDefaultControllerStarted,
writableStreamDefaultControllerStartFailed,
} = require('internal/webstreams/writablestream');
const assert = require('internal/assert');
const kSkipThrow = Symbol('kSkipThrow');
/**
* @typedef {import('./queuingstrategies').QueuingStrategy
* } QueuingStrategy
* @typedef {import('./queuingstrategies').QueuingStrategySize
* } QueuingStrategySize
*/
/**
* @callback TransformerStartCallback
* @param {TransformStreamDefaultController} controller
*/
/**
* @callback TransformerFlushCallback
* @param {TransformStreamDefaultController} controller
* @returns {Promise<void>}
*/
/**
* @callback TransformerTransformCallback
* @param {any} chunk
* @param {TransformStreamDefaultController} controller
* @returns {Promise<void>}
*/
/**
* @typedef {{
* start? : TransformerStartCallback,
* transform? : TransformerTransformCallback,
* flush? : TransformerFlushCallback,
* readableType? : any,
* writableType? : any,
* }} Transformer
*/
class TransformStream {
[kType] = 'TransformStream';
/**
* @param {Transformer} [transformer]
* @param {QueuingStrategy} [writableStrategy]
* @param {QueuingStrategy} [readableStrategy]
*/
constructor(
transformer = kEmptyObject,
writableStrategy = kEmptyObject,
readableStrategy = kEmptyObject) {
markTransferMode(this, false, true);
validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
const readableType = transformer?.readableType;
const writableType = transformer?.writableType;
const start = transformer?.start;
if (readableType !== undefined) {
throw new ERR_INVALID_ARG_VALUE.RangeError(
'transformer.readableType',
readableType);
}
if (writableType !== undefined) {
throw new ERR_INVALID_ARG_VALUE.RangeError(
'transformer.writableType',
writableType);
}
const readableHighWaterMark = readableStrategy?.highWaterMark;
const readableSize = readableStrategy?.size;
const writableHighWaterMark = writableStrategy?.highWaterMark;
const writableSize = writableStrategy?.size;
const actualReadableHighWaterMark =
extractHighWaterMark(readableHighWaterMark, 0);
const actualReadableSize = extractSizeAlgorithm(readableSize);
const actualWritableHighWaterMark =
extractHighWaterMark(writableHighWaterMark, 1);
const actualWritableSize = extractSizeAlgorithm(writableSize);
initializeTransformStream(
this,
actualWritableHighWaterMark,
actualWritableSize,
actualReadableHighWaterMark,
actualReadableSize);
setupTransformStreamDefaultControllerFromTransformer(this, transformer);
transformStreamStart(
this,
start === undefined ?
undefined :
FunctionPrototypeCall(start, transformer, this[kState].controller));
}
/**
* @readonly
* @type {ReadableStream}
*/
get readable() {
if (!isTransformStream(this))
throw new ERR_INVALID_THIS('TransformStream');
return this[kState].readable;
}
/**
* @readonly
* @type {WritableStream}
*/
get writable() {
if (!isTransformStream(this))
throw new ERR_INVALID_THIS('TransformStream');
return this[kState].writable;
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
readable: this.readable,
writable: this.writable,
backpressure: this[kState].backpressure,
});
}
[kTransfer]() {
if (!isTransformStream(this))
throw new ERR_INVALID_THIS('TransformStream');
const {
readable,
writable,
} = this[kState];
if (readable.locked) {
throw new DOMException(
'Cannot transfer a locked ReadableStream',
'DataCloneError');
}
if (writable.locked) {
throw new DOMException(
'Cannot transfer a locked WritableStream',
'DataCloneError');
}
return {
data: {
readable,
writable,
},
deserializeInfo:
'internal/webstreams/transformstream:TransferredTransformStream',
};
}
[kTransferList]() {
return [ this[kState].readable, this[kState].writable ];
}
[kDeserialize]({ readable, writable }) {
this[kState].readable = readable;
this[kState].writable = writable;
}
}
ObjectDefineProperties(TransformStream.prototype, {
readable: kEnumerableProperty,
writable: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(TransformStream.name),
});
// 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 TransformStreamState {
readable = undefined;
writable = undefined;
controller = undefined;
backpressure = undefined;
// Continuation slots replacing the spec's
// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
pullPending = false;
pendingWriteParked = false;
pendingWriteChunk = undefined;
pendingWriteContextFrame = undefined;
writeContinuation = undefined;
}
ObjectSetPrototypeOf(TransformStreamState.prototype, null);
class TransformStreamControllerState {
stream = undefined;
transformAlgorithm = undefined;
flushAlgorithm = undefined;
cancelAlgorithm = undefined;
performTransformRejected = undefined;
finishPromise = undefined;
}
ObjectSetPrototypeOf(TransformStreamControllerState.prototype, null);
function InternalTransferredTransformStream() {
ObjectSetPrototypeOf(this, TransformStream.prototype);
markTransferMode(this, false, true);
this[kType] = 'TransformStream';
this[kState] = new TransformStreamState();
}
ObjectSetPrototypeOf(InternalTransferredTransformStream.prototype, TransformStream.prototype);
ObjectSetPrototypeOf(InternalTransferredTransformStream, TransformStream);
function TransferredTransformStream() {
const stream = new InternalTransferredTransformStream();
stream.constructor = TransformStream;
return stream;
}
TransferredTransformStream.prototype[kDeserialize] = () => {};
class TransformStreamDefaultController {
[kType] = 'TransformStreamDefaultController';
[kState];
constructor(skipThrowSymbol = undefined) {
if (skipThrowSymbol !== kSkipThrow) {
throw new ERR_ILLEGAL_CONSTRUCTOR();
}
}
/**
* @readonly
* @type {number}
*/
get desiredSize() {
if (!isTransformStreamDefaultController(this))
throw new ERR_INVALID_THIS('TransformStreamDefaultController');
const {
stream,
} = this[kState];
const {
readable,
} = stream[kState];
const {
controller: readableController,
} = readable[kState];
return readableStreamDefaultControllerGetDesiredSize(readableController);
}
/**
* @param {any} [chunk]
*/
enqueue(chunk = undefined) {
if (!isTransformStreamDefaultController(this))
throw new ERR_INVALID_THIS('TransformStreamDefaultController');
transformStreamDefaultControllerEnqueue(this, chunk);
}
/**
* @param {any} [reason]
*/
error(reason = undefined) {
if (!isTransformStreamDefaultController(this))
throw new ERR_INVALID_THIS('TransformStreamDefaultController');
transformStreamDefaultControllerError(this, reason);
}
terminate() {
if (!isTransformStreamDefaultController(this))
throw new ERR_INVALID_THIS('TransformStreamDefaultController');
transformStreamDefaultControllerTerminate(this);
}
[kInspect](depth, options) {
return customInspect(depth, options, this[kType], {
stream: this[kState].stream,
});
}
}
ObjectDefineProperties(TransformStreamDefaultController.prototype, {
desiredSize: kEnumerableProperty,
enqueue: kEnumerableProperty,
error: kEnumerableProperty,
terminate: kEnumerableProperty,
[SymbolToStringTag]: getNonWritablePropertyDescriptor(TransformStreamDefaultController.name),
});
const isTransformStream =
isBrandCheck('TransformStream');
const isTransformStreamDefaultController =
isBrandCheck('TransformStreamDefaultController');
// Raw callback (see createRawCallback*): invoked inside the try/catch of
// transformStreamDefaultControllerPerformTransform.
function defaultTransformAlgorithm(chunk, controller) {
transformStreamDefaultControllerEnqueue(controller, chunk);
}
function initializeTransformStream(
stream,
writableHighWaterMark,
writableSizeAlgorithm,
readableHighWaterMark,
readableSizeAlgorithm) {
// The sink and source algorithms are shared functions that reach the
// transform stream through their controller, and the post-start step
// is delivered by transformStreamStart, so no per-stream closures or
// start promises are allocated here.
const writable = createWritableStream(
deferredStartAlgorithm,
transformStreamDefaultSinkWriteAlgorithm,
transformStreamDefaultSinkCloseAlgorithm,
transformStreamDefaultSinkAbortAlgorithm,
writableHighWaterMark,
writableSizeAlgorithm,
);
writable[kState].controller[kState].transformStream = stream;
const readable = createReadableStream(
deferredStartAlgorithm,
transformStreamDefaultSourcePullAlgorithm,
transformStreamDefaultSourceCancelAlgorithm,
readableHighWaterMark,
readableSizeAlgorithm,
);
readable[kState].controller[kState].transformStream = stream;
const state = new TransformStreamState();
state.readable = readable;
state.writable = writable;
stream[kState] = state;
transformStreamSetBackpressure(stream, true);
}
// Delivers the post-start step of both sides. The spec resolves the start
// promise with the transformer's start result at the end of construction
// and each side then adopts it through a wrapper promise, so with a
// non-thenable start result the started steps run three microtasks after
// construction, and with a thenable one two microtasks after the adopting
// promise settles. The same hops are taken as reactions on one shared
// promise instead of a promise record plus two wrapper promises.
function transformStreamStart(stream, startResult) {
const state = stream[kState];
const started = () => {
writableStreamDefaultControllerStarted(state.writable[kState].controller);
readableStreamDefaultControllerStarted(state.readable[kState].controller);
};
if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
PromisePrototypeThen(
PromisePrototypeThen(
PromisePrototypeThen(kResolvedPromise, undefined, undefined),
undefined,
undefined),
started);
return;
}
const startPromise = new Promise((resolve) => resolve(startResult));
PromisePrototypeThen(
PromisePrototypeThen(startPromise, undefined, undefined),
started,
(error) => {
writableStreamDefaultControllerStartFailed(
state.writable[kState].controller,
error);
readableStreamDefaultControllerError(
state.readable[kState].controller,
error);
});
}
function transformStreamError(stream, error) {
const {
readable,
} = stream[kState];
const {
controller,
} = readable[kState];
readableStreamDefaultControllerError(controller, error);
transformStreamErrorWritableAndUnblockWrite(stream, error);
}
function transformStreamErrorWritableAndUnblockWrite(stream, error) {
const {
controller,
writable,
} = stream[kState];
transformStreamDefaultControllerClearAlgorithms(controller);
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
error);
transformStreamUnblockWrite(stream);
}
function transformStreamUnblockWrite(stream) {
if (stream[kState].backpressure)
transformStreamSetBackpressure(stream, false);
}
// The spec's [[backpressureChangePromise]] is only ever observed by the
// source pull algorithm (settles when backpressure next becomes true) and
// by a sink write arriving while backpressure is set (settles when
// backpressure next becomes false). Both observers are internal, so the
// promise record is replaced by continuation slots: a parked pull is
// completed by delivering the readable controller's pull-fulfilled step,
// and a parked write by the cached write continuation (see
// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
// shared resolved promise at the exact microtask position the old
// record's reaction would have had.
function transformStreamSetBackpressure(stream, backpressure) {
const state = stream[kState];
assert(state.backpressure !== backpressure);
state.backpressure = backpressure;
if (backpressure) {
if (state.pullPending) {
state.pullPending = false;
// The pull-fulfilled step exists: a pull parked it (see
// transformStreamDefaultSourcePullAlgorithm), and the readable
// controller creates it before invoking the pull algorithm.
PromisePrototypeThen(
kResolvedPromise,
state.readable[kState].controller[kState].pullFulfilled);
}
} else if (state.pendingWriteParked) {
PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
}
}
function setupTransformStreamDefaultController(
stream,
controller,
transformAlgorithm,
flushAlgorithm,
cancelAlgorithm) {
assert(isTransformStream(stream));
assert(stream[kState].controller === undefined);
const controllerState = new TransformStreamControllerState();
controllerState.stream = stream;
controllerState.transformAlgorithm = transformAlgorithm;
controllerState.flushAlgorithm = flushAlgorithm;
controllerState.cancelAlgorithm = cancelAlgorithm;
controller[kState] = controllerState;
stream[kState].controller = controller;
}
function setupTransformStreamDefaultControllerFromTransformer(
stream,
transformer) {
const controller = new TransformStreamDefaultController(kSkipThrow);
const transform = transformer?.transform;
const flush = transformer?.flush;
const cancel = transformer?.cancel;
const transformAlgorithm = transform ?
createRawCallback2Params('transformer.transform', transform, transformer) :
defaultTransformAlgorithm;
const flushAlgorithm = flush ?
createPromiseCallback1Param('transformer.flush', flush, transformer) :
nonOpFlush;
const cancelAlgorithm = cancel ?
createPromiseCallback1Param('transformer.cancel', cancel, transformer) :
nonOpCancel;
setupTransformStreamDefaultController(
stream,
controller,
transformAlgorithm,
flushAlgorithm,
cancelAlgorithm);
}
function transformStreamDefaultControllerClearAlgorithms(controller) {
controller[kState].transformAlgorithm = undefined;
controller[kState].flushAlgorithm = undefined;
controller[kState].cancelAlgorithm = undefined;
}
function transformStreamDefaultControllerEnqueue(controller, chunk) {
const {
stream,
} = controller[kState];
const {
readable,
} = stream[kState];
const {
controller: readableController,
} = readable[kState];
if (!readableStreamDefaultControllerCanCloseOrEnqueue(readableController))
throw new ERR_INVALID_STATE.TypeError('Unable to enqueue');
try {
readableStreamDefaultControllerEnqueue(readableController, chunk);
} catch (error) {
transformStreamErrorWritableAndUnblockWrite(stream, error);
throw readable[kState].storedError;
}
const backpressure =
readableStreamDefaultControllerHasBackpressure(readableController);
if (backpressure !== stream[kState].backpressure) {
assert(backpressure);
transformStreamSetBackpressure(stream, true);
}
}
function transformStreamDefaultControllerError(controller, error) {
transformStreamError(controller[kState].stream, error);
}
// Mirrors the reference implementation's
// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
// the returned promise settles one microtask after the (coerced) result
// does, and a rejection errors the transform stream before propagating.
// The raw transform callback plus the shared resolved promise for
// non-thenable results replace the previous async wrapper's two implicit
// promises per chunk.
function transformStreamDefaultControllerPerformTransform(controller, chunk) {
const controllerState = controller[kState];
const transformAlgorithm = controllerState.transformAlgorithm;
if (transformAlgorithm === undefined) {
// Algorithms were cleared by a concurrent cancel/abort/close.
return kResolvedPromise;
}
let result;
try {
result = transformAlgorithm(chunk, controller);
} catch (error) {
result = PromiseReject(error);
}
if (result === null ||
(typeof result !== 'object' && typeof result !== 'function')) {
result = kResolvedPromise;
} else {
result = PromiseResolve(result);
}
controllerState.performTransformRejected ??= (error) => {
transformStreamError(controller[kState].stream, error);
throw error;
};
return PromisePrototypeThen(
result,
undefined,
controllerState.performTransformRejected);
}
function transformStreamDefaultControllerTerminate(controller) {
const {
stream,
} = controller[kState];
const {
readable,
} = stream[kState];
assert(readable !== undefined);
const {
controller: readableController,
} = readable[kState];
readableStreamDefaultControllerClose(readableController);
transformStreamErrorWritableAndUnblockWrite(
stream,
new ERR_INVALID_STATE.TypeError('TransformStream has been terminated'));
}
function transformStreamDefaultSinkWriteAlgorithm(chunk, writableController) {
const stream = writableController[kState].transformStream;
const state = stream[kState];
const {
writable,
controller,
} = state;
assert(writable[kState].state === 'writable');
if (state.backpressure) {
// Park the chunk; the backpressure -> false flip delivers the cached
// continuation (see transformStreamSetBackpressure) at the same
// microtask position as the old [[backpressureChangePromise]]
// reaction. The continuation completes the parked write by wiring
// the perform-transform promise directly to the writable
// controller's write reactions (they exist: the controller creates
// them before invoking the write algorithm), replacing the promise
// record the old code allocated and resolved per parked chunk. The
// writable dispatches a single write at a time, so one pending slot
// suffices. The old reaction was registered here, so the continuation
// runs in the async context captured at park time rather than in the
// context of the read that lifts the backpressure.
assert(!state.pendingWriteParked);
state.pendingWriteParked = true;
state.pendingWriteChunk = chunk;
state.pendingWriteContextFrame = AsyncContextFrame.current();
state.writeContinuation ??= () => {
const pendingChunk = state.pendingWriteChunk;
const prior = AsyncContextFrame.exchange(state.pendingWriteContextFrame);
state.pendingWriteParked = false;
state.pendingWriteChunk = undefined;
state.pendingWriteContextFrame = undefined;
try {
const writableState = state.writable[kState];
const writableControllerState = writableState.controller[kState];
if (writableState.state === 'erroring') {
const error = writableState.storedError;
PromisePrototypeThen(
kResolvedPromise,
() => writableControllerState.writeRejected(error));
return;
}
assert(writableState.state === 'writable');
PromisePrototypeThen(
transformStreamDefaultControllerPerformTransform(
controller,
pendingChunk),
writableControllerState.writeFulfilled,
writableControllerState.writeRejected);
} finally {
AsyncContextFrame.set(prior);
}
};
return kParkedAlgorithmResult;
}
return transformStreamDefaultControllerPerformTransform(controller, chunk);
}
async function transformStreamDefaultSinkAbortAlgorithm(
reason,
writableController) {
const stream = writableController[kState].transformStream;
const {
controller,
readable,
} = stream[kState];
if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}
const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const cancelPromise = controller[kState].cancelAlgorithm(reason);
transformStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
cancelPromise,
() => {
if (readable[kState].state === 'errored')
reject(readable[kState].storedError);
else {
readableStreamDefaultControllerError(readable[kState].controller, reason);
resolve();
}
},
(error) => {
readableStreamDefaultControllerError(readable[kState].controller, error);
reject(error);
},
);
return controller[kState].finishPromise;
}
function transformStreamDefaultSinkCloseAlgorithm(writableController) {
const stream = writableController[kState].transformStream;
const {
readable,
controller,
} = stream[kState];
if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}
const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const flushPromise = controller[kState].flushAlgorithm(controller);
transformStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
flushPromise,
() => {
if (readable[kState].state === 'errored')
reject(readable[kState].storedError);
else {
readableStreamDefaultControllerClose(readable[kState].controller);
resolve();
}
},
(error) => {
readableStreamDefaultControllerError(readable[kState].controller, error);
reject(error);
});
return controller[kState].finishPromise;
}
function transformStreamDefaultSourcePullAlgorithm(readableController) {
const stream = readableController[kState].transformStream;
const state = stream[kState];
assert(state.backpressure);
transformStreamSetBackpressure(stream, false);
// Park the pull: the next backpressure -> true flip delivers the
// pull-fulfilled step (see transformStreamSetBackpressure). The old
// [[backpressureChangePromise]] this replaces was only ever resolved,
// so the parked pull needs no rejection delivery.
state.pullPending = true;
return kParkedAlgorithmResult;
}
function transformStreamDefaultSourceCancelAlgorithm(
reason,
readableController) {
const stream = readableController[kState].transformStream;
const {
controller,
writable,
} = stream[kState];
if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}
const { promise, resolve, reject } = PromiseWithResolvers();
controller[kState].finishPromise = promise;
const cancelPromise = controller[kState].cancelAlgorithm(reason);
transformStreamDefaultControllerClearAlgorithms(controller);
PromisePrototypeThen(
cancelPromise,
() => {
if (writable[kState].state === 'errored')
reject(writable[kState].storedError);
else {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
reason);
transformStreamUnblockWrite(stream);
resolve();
}
},
(error) => {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
error);
transformStreamUnblockWrite(stream);
reject(error);
},
);
return controller[kState].finishPromise;
}
module.exports = {
TransformStream,
TransformStreamDefaultController,
TransferredTransformStream,
// Exported Brand Checks
isTransformStream,
isTransformStreamDefaultController,
};