'use strict';
const {
Array,
ArrayBufferPrototypeGetByteLength,
ArrayBufferPrototypeGetDetached,
ArrayBufferPrototypeGetResizable,
ArrayPrototypePush,
ArrayPrototypeSlice,
ObjectDefineProperty,
ObjectFreeze,
PromiseResolve,
PromiseWithResolvers,
SafePromisePrototypeFinally,
SafePromiseRace,
SafeWeakSet,
SymbolAsyncIterator,
TypedArrayPrototypeGetBuffer,
TypedArrayPrototypeGetByteLength,
TypedArrayPrototypeGetByteOffset,
TypedArrayPrototypeSet,
Uint8Array,
} = primordials;
const {
markPromiseAsHandled,
} = internalBinding('util');
const { TextEncoder } = require('internal/encoding');
const {
codes: {
ERR_INVALID_ARG_TYPE,
ERR_INVALID_STATE,
},
} = require('internal/errors');
const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types');
const {
validateInteger,
validateOneOf,
} = require('internal/validators');
const {
converters,
} = require('internal/streams/iter/webidl');
const {
kValidatedTransform,
} = require('internal/streams/iter/types');
// Shared `addEventListener()` options for one-time 'abort' listeners. Frozen
// because signals can come from user code, and a patched addEventListener()
// must not be able to change the options for every later registration.
const kNullOnceOption = ObjectFreeze({ __proto__: null, once: true });
// Cached resolved promise to avoid allocating a new one on every sync fast-path.
const kResolvedPromise = PromiseResolve();
// Shared TextEncoder instance for string conversion.
const encoder = new TextEncoder();
// Default high water marks for push and multi-consumer streams. These values
// are somewhat arbitrary but have been tested across various workloads and
// appear to yield the best overall throughput/latency balance.
/** Minimum and default byte budget for push streams (single-consumer). */
const kPushDefaultBudget = 16384;
/** Default byte budget for broadcast and share streams (multi-consumer). */
const kMultiConsumerDefaultBudget = 65536;
/**
* Iterator result object (`{ done, value }`) for the iterators returned by
* this module. A result is created for every chunk, so it must be cheap.
*
* `new IterResult(done, value)` literals are created in V8 dictionary
* mode, which costs several times more than an ordinary object. Instances
* of this constructor have fast properties, always with the same shape
* (`done` before `value`), and still have no %Object.prototype% in their
* prototype chain, so a polluted `Object.prototype.then` cannot turn a
* result into a thenable when an async `next()` resolves with it. The
* prototype is a single empty, frozen, null-prototype object (V8 gives
* objects fast properties once they are used as a prototype).
* @param {boolean} done
* @param {any} value
*/
function IterResult(done, value) {
this.done = done;
this.value = value;
}
// The non-enumerable `constructor` lets util.inspect() print results as
// `IterResult { done, value }`.
IterResult.prototype = ObjectFreeze(ObjectDefineProperty(
{ __proto__: null }, 'constructor', { __proto__: null, value: IterResult }));
/**
* Register a handler for an AbortSignal, handling the already-aborted case.
* If the signal is already aborted, calls handler immediately.
* Otherwise, adds a one-time 'abort' listener.
* @param {AbortSignal} signal
* @param {Function} handler
*/
function onSignalAbort(signal, handler) {
if (signal.aborted) {
handler();
} else {
signal.addEventListener('abort', handler, kNullOnceOption);
}
}
function getOnAbort(reject, signal) {
return () => reject(signal.reason);
}
/**
* Read one item from an async iterator, rejecting early if the signal aborts.
* @param {AsyncIterator} iterator - The iterator to read from.
* @param {AbortSignal|undefined} signal - Optional abort signal.
* @returns {Promise<IteratorResult<Uint8Array[]>>|IteratorResult<Uint8Array[]>}
*/
function abortableNext(iterator, signal) {
if (signal === undefined) {
return iterator.next();
}
signal.throwIfAborted();
const next = iterator.next();
const { promise, reject } = PromiseWithResolvers();
const onAbort = getOnAbort(reject, signal);
signal.addEventListener('abort', onAbort, kNullOnceOption);
if (signal.aborted) {
onAbort();
}
return SafePromisePrototypeFinally(SafePromiseRace([next, promise]), () => {
signal.removeEventListener('abort', onAbort);
});
}
/**
* Wrap an async source so each pending read is abort-aware.
* @param {AsyncIterable<Uint8Array[]>} source - The source to read from.
* @param {AbortSignal|undefined} signal - Optional abort signal.
* @returns {AsyncIterable<Uint8Array[]>}
*/
function yieldAbortable(source, signal) {
if (signal === undefined) {
return source;
}
return {
__proto__: null,
async *[SymbolAsyncIterator]() {
const iterator = source[SymbolAsyncIterator]();
let completed = false;
let aborted = false;
try {
while (true) {
const { done, value } = await abortableNext(iterator, signal);
if (done) {
completed = true;
return;
}
signal.throwIfAborted();
yield value;
}
} catch (error) {
aborted = signal.aborted;
throw error;
} finally {
if (!completed && typeof iterator.return === 'function') {
const result = iterator.return();
if (aborted) {
// PromiseResolve(result) can reject if result is a thenable that
// rejects, so mark it as handled even though the abort takes
// precedence over the result of iterator.return().
markPromiseAsHandled(PromiseResolve(result));
} else {
await result;
}
}
}
},
};
}
/**
* Compute the minimum cursor across a set of consumers and count how many
* consumers are at that cursor.
* @param {Set} consumers - Set of objects with a `cursor` property
* @param {number} fallback - Cursor to return when set is empty
* @returns {{ minCursor: number, minCursorConsumers: number }}
*/
function getMinCursor(consumers, fallback) {
let minCursor = fallback;
let minCursorConsumers = 0;
for (const consumer of consumers) {
if (consumer.cursor < minCursor) {
minCursor = consumer.cursor;
minCursorConsumers = 1;
} else if (consumer.cursor === minCursor) {
minCursorConsumers++;
}
}
return { __proto__: null, minCursor, minCursorConsumers };
}
/**
* Convert a chunk (string or Uint8Array) to Uint8Array.
* Strings are UTF-8 encoded.
* @param {Uint8Array|string} chunk
* @returns {Uint8Array}
*/
function toUint8Array(chunk) {
if (typeof chunk === 'string') {
return encoder.encode(chunk);
}
if (!isUint8Array(chunk)) {
throw new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Uint8Array'], chunk);
}
return chunk;
}
// Byte view snapshots and batch entries are created for every chunk and
// every batch, so like IterResult they are constructed rather than created
// as `{ __proto__: null, ... }` literals (which are dictionary-mode objects),
// and their prototype is an empty null-prototype object.
function ByteViewSnapshot(value, buffer, sharedBufferView) {
this.value = value;
this.buffer = buffer;
this.bufferByteLength = sharedBufferView === undefined ?
ArrayBufferPrototypeGetByteLength(buffer) :
TypedArrayPrototypeGetByteLength(sharedBufferView);
this.byteLength = TypedArrayPrototypeGetByteLength(value);
this.byteOffset = TypedArrayPrototypeGetByteOffset(value);
this.detached = sharedBufferView === undefined &&
ArrayBufferPrototypeGetDetached(buffer);
this.sharedBufferView = sharedBufferView;
}
ByteViewSnapshot.prototype = ObjectFreeze({ __proto__: null });
function BatchEntry(views, byteLength) {
this.views = views;
this.byteLength = byteLength;
}
BatchEntry.prototype = ObjectFreeze({ __proto__: null });
// Waiters for reads, writes and drains are queued whenever a stream has
// to wait, which can be once per chunk.
function PendingRequest(resolve, reject) {
this.resolve = resolve;
this.reject = reject;
}
PendingRequest.prototype = ObjectFreeze({ __proto__: null });
function PendingWrite(batch, resolve, reject) {
this.batch = batch;
this.resolve = resolve;
this.reject = reject;
}
PendingWrite.prototype = ObjectFreeze({ __proto__: null });
function snapshotByteView(value) {
const buffer = TypedArrayPrototypeGetBuffer(value);
const sharedBufferView = isSharedArrayBuffer(buffer) ?
new Uint8Array(buffer) : undefined;
return new ByteViewSnapshot(value, buffer, sharedBufferView);
}
function validateByteView(snapshot) {
const {
value,
buffer,
bufferByteLength,
byteLength,
byteOffset,
detached,
sharedBufferView,
} = snapshot;
const currentBufferByteLength = sharedBufferView === undefined ?
ArrayBufferPrototypeGetByteLength(buffer) :
TypedArrayPrototypeGetByteLength(sharedBufferView);
const currentDetached = sharedBufferView === undefined &&
ArrayBufferPrototypeGetDetached(buffer);
if (TypedArrayPrototypeGetBuffer(value) !== buffer ||
currentBufferByteLength !== bufferByteLength ||
TypedArrayPrototypeGetByteLength(value) !== byteLength ||
TypedArrayPrototypeGetByteOffset(value) !== byteOffset ||
currentDetached !== detached) {
throw new ERR_INVALID_STATE.TypeError(
'Byte view was resized or detached after being accepted');
}
return value;
}
/**
* Validate an explicit `budget` option. The spec only requires the default
* budget to be at least 16384 bytes; any positive explicit budget is valid.
* @param {unknown} budget
*/
function validateBudget(budget) {
validateInteger(budget, 'options.budget', 1);
}
/**
* Append a chunk to `chunks` for later concatenation, and the information
* needed to detect that it was resized or detached in the meantime to
* `checks`. This provides the same guarantee as snapshotByteView() and
* validateByteView() without allocating a snapshot for every chunk in the
* common case: a non-empty view of a fixed-length, non-shared ArrayBuffer can
* only change by the buffer being detached, which makes its byteLength 0, so
* its byteLength is all that needs to be recorded.
* @param {Uint8Array[]} chunks
* @param {Array<number|object>} checks
* @param {Uint8Array} value
* @returns {number} The byteLength of `value`.
*/
function recordChunk(chunks, checks, value) {
const buffer = TypedArrayPrototypeGetBuffer(value);
const byteLength = TypedArrayPrototypeGetByteLength(value);
ArrayPrototypePush(chunks, value);
if (byteLength === 0 || isSharedArrayBuffer(buffer) ||
ArrayBufferPrototypeGetResizable(buffer)) {
ArrayPrototypePush(checks, snapshotByteView(value));
} else {
ArrayPrototypePush(checks, byteLength);
}
return byteLength;
}
/**
* Validate chunks recorded with recordChunk().
* @param {Uint8Array[]} chunks
* @param {Array<number|object>} checks
* @returns {Uint8Array[]} `chunks`
*/
function validateRecordedChunks(chunks, checks) {
for (let i = 0; i < chunks.length; i++) {
const check = checks[i];
if (typeof check !== 'number') {
validateByteView(check);
} else if (TypedArrayPrototypeGetByteLength(chunks[i]) !== check) {
throw new ERR_INVALID_STATE.TypeError(
'Byte view was resized or detached after being accepted');
}
}
return chunks;
}
function createBatchEntry(chunks) {
const views = new Array(chunks.length);
let byteLength = 0;
for (let i = 0; i < chunks.length; i++) {
const view = snapshotByteView(chunks[i]);
views[i] = view;
byteLength += view.byteLength;
}
return new BatchEntry(views, byteLength);
}
/**
* Split a batch entry into consecutive entries that are each smaller than
* `limit` bytes, preserving chunk order. A single chunk of `limit` bytes or
* more forms an entry on its own. Returns `undefined` if no split is needed.
* @param {{ views: object[], byteLength: number }} entry
* @param {number} limit
* @returns {Array<{ views: object[], byteLength: number }>|undefined}
*/
function splitBatchEntry(entry, limit) {
const { views } = entry;
if (entry.byteLength < limit || views.length < 2) return undefined;
const entries = [];
let current = [];
let byteLength = 0;
for (let i = 0; i < views.length; i++) {
const view = views[i];
if (current.length > 0 && byteLength + view.byteLength >= limit) {
ArrayPrototypePush(entries, new BatchEntry(current, byteLength));
current = [];
byteLength = 0;
}
ArrayPrototypePush(current, view);
byteLength += view.byteLength;
}
ArrayPrototypePush(entries, new BatchEntry(current, byteLength));
return entries;
}
function validateBatchEntry(entry) {
const chunks = new Array(entry.views.length);
for (let i = 0; i < entry.views.length; i++) {
chunks[i] = validateByteView(entry.views[i]);
}
return chunks;
}
function copyBytes(chunk) {
const copy = new Uint8Array(TypedArrayPrototypeGetByteLength(chunk));
TypedArrayPrototypeSet(copy, chunk);
return copy;
}
/**
* Concatenate multiple Uint8Arrays into a single Uint8Array.
* @param {Uint8Array[]} chunks
* @returns {Uint8Array}
*/
function concatBytes(chunks) {
// Empty stream: return zero-length Uint8Array
if (chunks.length === 0) {
return new Uint8Array(0);
}
// Single chunk: return directly if it covers the entire backing buffer,
// otherwise return a copy
if (chunks.length === 1) {
const chunk = chunks[0];
// If non-zero offset, skip the remaining buffer checks.
if (TypedArrayPrototypeGetByteOffset(chunk) === 0) {
const buf = TypedArrayPrototypeGetBuffer(chunk);
if (
!isSharedArrayBuffer(buf) &&
TypedArrayPrototypeGetByteLength(chunk) ===
ArrayBufferPrototypeGetByteLength(buf)
) {
return chunk;
}
}
return copyBytes(chunk);
}
// Multiple chunks: concatenate
let totalByteLength = 0;
for (let i = 0; i < chunks.length; i++) {
totalByteLength += TypedArrayPrototypeGetByteLength(chunks[i]);
}
const concatenated = new Uint8Array(totalByteLength);
let offset = 0;
for (let i = 0; i < chunks.length; i++) {
TypedArrayPrototypeSet(concatenated, chunks[i], offset);
offset += TypedArrayPrototypeGetByteLength(chunks[i]);
}
return concatenated;
}
// Conversion contexts for the per-write paths. The converters only read
// them (to build error messages), so they are shared rather than allocated
// on every write.
const kChunkContext = ObjectFreeze({ __proto__: null, context: 'chunk' });
const kChunksContext = ObjectFreeze({ __proto__: null, context: 'chunks' });
const kWriteOptionsContext =
ObjectFreeze({ __proto__: null, context: 'options' });
/**
* Convert an array of chunks (strings or Uint8Arrays) to a Uint8Array[].
* Always returns a fresh copy of the array.
* @param {Array<Uint8Array|string>} chunks
* @returns {Uint8Array[]}
*/
function convertChunks(chunks) {
chunks = converters.WriterChunkSequence(chunks, kChunksContext);
const len = chunks.length;
const result = new Array(len);
for (let i = 0; i < len; i++) {
result[i] = toUint8Array(chunks[i]);
}
return result;
}
/**
* Validate Writer options and return options.signal.
* @param {object|undefined} options
* @returns {AbortSignal|undefined}
*/
function getWriterSignal(options) {
return converters.WriteOptions(options, kWriteOptionsContext).signal;
}
function toWriterUint8Array(chunk) {
return toUint8Array(converters.WriterChunk(chunk, kChunkContext));
}
/**
* Check if a value implements a Symbol-keyed protocol (has a function
* at the given symbol key).
* @param {unknown} value
* @param {symbol} symbol
* @returns {boolean}
*/
function getProtocolMethod(value, symbol) {
// Functions are objects too, and may implement the protocols.
if (value === null ||
(typeof value !== 'object' && typeof value !== 'function') ||
!(symbol in value)) {
return undefined;
}
const method = value[symbol];
return typeof method === 'function' ? method : undefined;
}
function hasProtocol(value, symbol) {
return getProtocolMethod(value, symbol) !== undefined;
}
const transformRecords = new SafeWeakSet();
/**
* Read and retain a stateful transform's callable exactly once.
* @param {unknown} value
* @returns {Function|object|undefined}
*/
function snapshotTransform(value) {
if (typeof value === 'function') return value;
if (transformRecords.has(value)) return value;
const transform = value?.transform;
if (typeof transform !== 'function') return undefined;
const record = {
__proto__: null,
transform,
receiver: value,
[kValidatedTransform]: value[kValidatedTransform],
};
transformRecords.add(record);
return record;
}
/**
* Check if a value is a stateful transform object (has a transform method).
* @param {unknown} value
* @returns {boolean}
*/
function isTransformObject(value) {
return transformRecords.has(value) || typeof value?.transform === 'function';
}
/**
* Check if a value is a valid transform (function or transform object).
* @param {unknown} value
* @returns {boolean}
*/
function isTransform(value) {
return typeof value === 'function' || isTransformObject(value);
}
/**
* Parse variadic arguments for pull/pullSync.
* Returns { transforms, options }
* @param {Array} args
* @returns {{ transforms: Array, options: unknown }}
*/
function parsePullArgs(args) {
if (args.length === 0) {
return { __proto__: null, transforms: [], options: undefined };
}
let transforms;
let options;
const last = args[args.length - 1];
const lastTransform = snapshotTransform(last);
if (lastTransform === undefined) {
transforms = ArrayPrototypeSlice(args, 0, -1);
options = last;
} else {
transforms = ArrayPrototypeSlice(args);
transforms[transforms.length - 1] = lastTransform;
options = undefined;
}
for (let i = 0; i < transforms.length; i++) {
const transform = snapshotTransform(transforms[i]);
if (transform === undefined) {
throw new ERR_INVALID_ARG_TYPE(
`transforms[${i}]`, ['Function', 'Object with transform()'],
transforms[i]);
}
transforms[i] = transform;
}
return { __proto__: null, transforms, options };
}
/**
* Validate backpressure option value.
* @param {string} value
*/
function validateBackpressure(value) {
validateOneOf(value, 'options.backpressure', [
'strict',
'unbounded',
'drop-oldest',
'drop-newest',
]);
}
module.exports = {
IterResult,
PendingRequest,
PendingWrite,
kMultiConsumerDefaultBudget,
kNullOnceOption,
kPushDefaultBudget,
kResolvedPromise,
concatBytes,
convertChunks,
createBatchEntry,
recordChunk,
splitBatchEntry,
getProtocolMethod,
getWriterSignal,
getMinCursor,
hasProtocol,
isTransform,
isTransformObject,
onSignalAbort,
parsePullArgs,
snapshotTransform,
toUint8Array,
toWriterUint8Array,
validateBackpressure,
validateBatchEntry,
validateBudget,
validateRecordedChunks,
validateByteView,
yieldAbortable,
};