'use strict';
// New Streams API - Share
//
// Pull-model multi-consumer streaming. Shares a single source among
// multiple consumers with explicit buffering.
const {
ArrayPrototypePush,
FunctionPrototypeCall,
ObjectSetPrototypeOf,
PromisePrototypeThen,
PromiseResolve,
PromiseWithResolvers,
SafePromiseRace,
SafeSet,
Symbol,
SymbolAsyncIterator,
SymbolDispose,
SymbolIterator,
} = primordials;
const {
shareProtocol,
shareSyncProtocol,
} = require('internal/streams/iter/types');
const {
from,
fromSync,
isAsyncIterable,
isSyncIterable,
} = require('internal/streams/iter/from');
const {
pullSync: pullSyncWithTransforms,
pullWithConsumerCleanup,
} = require('internal/streams/iter/pull');
const {
IterResult,
kMultiConsumerDefaultBudget,
createBatchEntry,
getProtocolMethod,
getMinCursor,
onSignalAbort,
parsePullArgs,
splitBatchEntry,
validateBatchEntry,
validateBudget,
} = require('internal/streams/iter/utils');
const {
converters,
} = require('internal/streams/iter/webidl');
const {
RingBuffer,
} = require('internal/streams/iter/ringbuffer');
const {
codes: {
ERR_INVALID_ARG_TYPE,
ERR_INVALID_ARG_VALUE,
ERR_INVALID_RETURN_VALUE,
ERR_INVALID_STATE,
ERR_OUT_OF_RANGE,
},
} = require('internal/errors');
const { markPromiseAsHandled } = internalBinding('util');
// =============================================================================
// Async Share Implementation
// =============================================================================
const kNoShareError = Symbol('kNoShareError');
const kShareCancelled = Symbol('kShareCancelled');
const kSetFactorySignal = Symbol('kSetFactorySignal');
class ShareImpl {
#source;
#options;
#buffer = new RingBuffer();
#bufferStart = 0;
#consumers = new SafeSet();
#sourceIterator = null;
#sourceExhausted = false;
#sourceError = kNoShareError;
#cancelled = false;
#pulling = false;
#pullWaiters = [];
#cancelPromise;
#resolveCancel;
#cancelError = kNoShareError;
#cachedMinCursor = 0;
#cachedMinCursorConsumers = 0;
#abortHandler;
/** Cumulative byte size of buffered entries */
#bufferedBytes = 0;
constructor(source, options) {
this.#source = source;
this.#options = options;
const { promise, resolve } = PromiseWithResolvers();
this.#cancelPromise = promise;
this.#resolveCancel = resolve;
}
get consumerCount() {
return this.#consumers.size;
}
[kSetFactorySignal](signal) {
this.#abortHandler = () => this.cancel(signal.reason);
onSignalAbort(signal, this.#abortHandler);
}
pull(...args) {
const parsed = parsePullArgs(args);
const { transforms } = parsed;
const options = converters.PullOptions(parsed.options, {
__proto__: null,
context: 'options',
});
const { signal } = options;
// Avoid registering a consumer that the pre-aborted pipeline will never
// read or detach.
if (signal?.aborted) {
return {
__proto__: null,
// eslint-disable-next-line require-yield
async *[SymbolAsyncIterator]() {
throw signal.reason;
},
};
}
const rawConsumer = this.#createRawConsumer();
if (transforms.length > 0 || signal) {
return pullWithConsumerCleanup(rawConsumer, transforms, signal);
}
return rawConsumer;
}
#createRawConsumer() {
const state = ObjectSetPrototypeOf({
cursor: this.#bufferStart,
resolve: null,
reject: null,
detached: false,
error: kNoShareError,
pendingNext: PromiseResolve(),
}, null);
this.#consumers.add(state);
if (this.#consumers.size === 1) {
this.#cachedMinCursor = state.cursor;
this.#cachedMinCursorConsumers = 1;
} else if (state.cursor === this.#cachedMinCursor) {
this.#cachedMinCursorConsumers++;
} else {
this.#recomputeMinCursor();
}
const self = this;
return {
__proto__: null,
[SymbolAsyncIterator]() {
const getNext = async () => {
// Loop until we get data, source is exhausted, or
// consumer is detached. Multiple consumers may be woken
// after a single pull - those that find no data at their
// cursor must re-pull rather than terminating prematurely.
for (;;) {
if (state.detached) {
if (state.error !== kNoShareError) throw state.error;
return new IterResult(true, undefined);
}
if (self.#cancelled) {
state.detached = true;
state.error = self.#cancelError;
self.#deleteConsumer(state);
if (state.error !== kNoShareError) throw state.error;
return new IterResult(true, undefined);
}
// Check if data is available in buffer
const bufferIndex = state.cursor - self.#bufferStart;
if (bufferIndex < self.#buffer.length) {
const chunk = self.#readEntry(self.#buffer.get(bufferIndex));
const cursor = state.cursor;
state.cursor++;
if (cursor === self.#cachedMinCursor &&
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
state.detached = true;
self.#deleteConsumer(state);
if (self.#sourceError !== kNoShareError) {
state.error = self.#sourceError;
throw state.error;
}
return new IterResult(true, undefined);
}
// Need to pull from source - check buffer limit
let shouldBuffer;
try {
shouldBuffer = await self.#waitForBufferSpace();
} catch (error) {
state.detached = true;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
throw error;
}
if (shouldBuffer === null) {
state.detached = true;
state.error = self.#cancelError;
self.#deleteConsumer(state);
if (state.error !== kNoShareError) throw state.error;
return new IterResult(true, undefined);
}
await self.#pullFromSource(!shouldBuffer);
if (!shouldBuffer) {
await self.#waitForBufferSpaceAfterDrop();
}
}
};
return ObjectSetPrototypeOf({
next() {
const next = PromisePrototypeThen(
state.pendingNext,
getNext,
getNext);
state.pendingNext = next;
markPromiseAsHandled(state.pendingNext);
return next;
},
async return() {
state.detached = true;
state.resolve = null;
state.reject = null;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
return new IterResult(true, undefined);
},
async throw() {
state.detached = true;
state.resolve = null;
state.reject = null;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
return new IterResult(true, undefined);
},
}, null);
},
};
}
cancel(reason) {
if (this.#cancelled) return;
const hasReason = arguments.length > 0;
this.#cancelled = true;
if (hasReason) {
this.#cancelError = reason;
}
this.#resolveCancel(kShareCancelled);
this.#resolveCancel = null;
try {
const returnMethod = this.#sourceIterator?.return;
if (typeof returnMethod === 'function') {
const cleanup = PromisePrototypeThen(
PromiseResolve(FunctionPrototypeCall(
returnMethod, this.#sourceIterator)),
undefined,
undefined);
markPromiseAsHandled(cleanup);
}
} catch {
// Cancellation has precedence over source cleanup errors.
}
for (const consumer of this.#consumers) {
consumer.error = this.#cancelError;
if (consumer.resolve) {
if (hasReason) {
consumer.reject?.(reason);
} else {
consumer.resolve(new IterResult(true, undefined));
}
consumer.resolve = null;
consumer.reject = null;
}
consumer.detached = true;
}
this.#consumers.clear();
this.#buffer.clear();
this.#bufferedBytes = 0;
this.#cleanupFactorySignal();
for (let i = 0; i < this.#pullWaiters.length; i++) {
this.#pullWaiters[i]();
}
this.#pullWaiters = [];
}
[SymbolDispose]() {
this.cancel();
}
// Internal methods
async #waitForBufferSpace() {
while (this.#bufferedBytes >= this.#options.budget) {
if (this.#cancelled ||
this.#sourceError !== kNoShareError ||
this.#sourceExhausted) {
return this.#cancelled ? null : true;
}
switch (this.#options.backpressure) {
case 'strict':
throw new ERR_OUT_OF_RANGE(
'buffered bytes', `< ${this.#options.budget}`,
this.#bufferedBytes);
case 'unbounded': {
const { promise, resolve } = PromiseWithResolvers();
ArrayPrototypePush(this.#pullWaiters, resolve);
await promise;
break;
}
case 'drop-oldest':
while (this.#bufferedBytes >= this.#options.budget &&
this.#buffer.length > 0) {
const evicted = this.#buffer.shift();
this.#bufferedBytes -= evicted.byteLength;
this.#bufferStart++;
}
for (const consumer of this.#consumers) {
if (consumer.cursor < this.#bufferStart) {
this.#deleteConsumerFromMin(consumer);
consumer.cursor = this.#bufferStart;
}
}
this.#recomputeMinCursor();
return true;
case 'drop-newest':
return false;
}
}
return true;
}
async #waitForBufferSpaceAfterDrop() {
while (this.#bufferedBytes >= this.#options.budget &&
!this.#cancelled &&
this.#sourceError === kNoShareError &&
!this.#sourceExhausted) {
const { promise, resolve } = PromiseWithResolvers();
ArrayPrototypePush(this.#pullWaiters, resolve);
await promise;
}
}
#pullFromSource(discard = false) {
if (this.#sourceExhausted || this.#cancelled) {
return PromiseResolve();
}
if (this.#pulling) {
const { promise, resolve } = PromiseWithResolvers();
ArrayPrototypePush(this.#pullWaiters, resolve);
return promise;
}
this.#pulling = true;
return (async () => {
try {
if (!this.#sourceIterator) {
if (isAsyncIterable(this.#source)) {
this.#sourceIterator =
this.#source[SymbolAsyncIterator]();
} else if (isSyncIterable(this.#source)) {
const syncIterator =
this.#source[SymbolIterator]();
this.#sourceIterator = ObjectSetPrototypeOf({
async next() {
return syncIterator.next();
},
async return() {
return syncIterator.return?.() ??
new IterResult(true, undefined);
},
}, null);
} else {
throw new ERR_INVALID_ARG_TYPE(
'source', ['AsyncIterable', 'Iterable'], this.#source);
}
}
const result = await SafePromiseRace([
this.#sourceIterator.next(),
this.#cancelPromise,
]);
if (this.#cancelled || result === kShareCancelled) return;
if (result.done) {
this.#sourceExhausted = true;
} else if (!discard) {
this.#bufferBatch(result.value);
}
} catch (error) {
this.#sourceError = error;
this.#sourceExhausted = true;
} finally {
if (this.#sourceExhausted && this.#consumers.size === 0) {
this.#cleanupFactorySignal();
}
this.#pulling = false;
for (let i = 0; i < this.#pullWaiters.length; i++) {
this.#pullWaiters[i]();
}
this.#pullWaiters = [];
}
})();
}
#bufferBatch(batch) {
const entry = createBatchEntry(batch);
// 'drop-oldest' evicts whole entries. A single pulled batch can be much
// larger than the budget (for example when from() combines many values
// of a sync source), and evicting it would discard every chunk in it,
// including ones that no consumer has read. Split such batches so that
// eviction keeps the newest chunks that fit the budget.
const entries = this.#options.backpressure === 'drop-oldest' ?
splitBatchEntry(entry, this.#options.budget) : undefined;
if (entries === undefined) {
this.#buffer.push(entry);
} else {
for (let i = 0; i < entries.length; i++) {
this.#buffer.push(entries[i]);
}
}
this.#bufferedBytes += entry.byteLength;
}
#tryTrimBuffer() {
// Retain buffered data for consumers that attach while none are active.
// Without this, the last consumer detaching would discard entries that
// a late-joining consumer has not read and cannot get back from the
// source.
if (this.#consumers.size === 0) return;
if (this.#cachedMinCursorConsumers === 0) {
this.#recomputeMinCursor();
}
const trimCount = this.#cachedMinCursor - this.#bufferStart;
if (trimCount > 0) {
for (let i = 0; i < trimCount; i++) {
const evicted = this.#buffer.get(i);
this.#bufferedBytes -= evicted.byteLength;
}
this.#buffer.trimFront(trimCount);
this.#bufferStart = this.#cachedMinCursor;
for (let i = 0; i < this.#pullWaiters.length; i++) {
this.#pullWaiters[i]();
}
this.#pullWaiters = [];
}
}
#readEntry(entry) {
try {
return validateBatchEntry(entry);
} catch (error) {
this.cancel(error);
this.#buffer.clear();
this.#bufferedBytes = 0;
throw error;
}
}
#recomputeMinCursor() {
const { minCursor, minCursorConsumers } = getMinCursor(
this.#consumers, this.#bufferStart + this.#buffer.length);
this.#cachedMinCursor = minCursor;
this.#cachedMinCursorConsumers = minCursorConsumers;
}
#cleanupFactorySignal() {
if (this.#abortHandler !== undefined) {
this.#options.signal.removeEventListener('abort', this.#abortHandler);
this.#abortHandler = undefined;
}
}
#deleteConsumerFromMin(consumer) {
if (consumer.cursor === this.#cachedMinCursor) {
this.#cachedMinCursorConsumers--;
return this.#cachedMinCursorConsumers === 0;
}
return false;
}
#deleteConsumer(consumer) {
if (this.#consumers.delete(consumer)) {
const shouldTrimBuffer = this.#deleteConsumerFromMin(consumer);
if (this.#sourceExhausted && this.#consumers.size === 0) {
this.#cleanupFactorySignal();
}
return shouldTrimBuffer;
}
return false;
}
}
// =============================================================================
// Sync Share Implementation
// =============================================================================
class SyncShareImpl {
#source;
#options;
#buffer = new RingBuffer();
#bufferStart = 0;
#consumers = new SafeSet();
#sourceIterator = null;
#sourceExhausted = false;
#sourceError = kNoShareError;
#cancelled = false;
#pulling = false;
#cachedMinCursor = 0;
#cachedMinCursorConsumers = 0;
/** Cumulative byte size of buffered entries */
#bufferedBytes = 0;
constructor(source, options) {
this.#source = source;
this.#options = options;
}
get consumerCount() {
return this.#consumers.size;
}
pull(...transforms) {
const rawConsumer = this.#createRawConsumer();
if (transforms.length > 0) {
return pullSyncWithTransforms(rawConsumer, ...transforms);
}
return rawConsumer;
}
#createRawConsumer() {
const state = ObjectSetPrototypeOf({
cursor: this.#bufferStart,
detached: false,
error: kNoShareError,
}, null);
this.#consumers.add(state);
if (this.#consumers.size === 1) {
this.#cachedMinCursor = state.cursor;
this.#cachedMinCursorConsumers = 1;
} else if (state.cursor === this.#cachedMinCursor) {
this.#cachedMinCursorConsumers++;
} else {
this.#recomputeMinCursor();
}
const self = this;
return {
__proto__: null,
[SymbolIterator]() {
return ObjectSetPrototypeOf({
next() {
if (state.detached) {
if (state.error !== kNoShareError) throw state.error;
return new IterResult(true, undefined);
}
if (self.#sourceError !== kNoShareError) {
state.detached = true;
state.error = self.#sourceError;
self.#deleteConsumer(state);
throw state.error;
}
if (self.#cancelled) {
state.detached = true;
self.#deleteConsumer(state);
return new IterResult(true, undefined);
}
const bufferIndex = state.cursor - self.#bufferStart;
if (bufferIndex < self.#buffer.length) {
const chunk = self.#readEntry(self.#buffer.get(bufferIndex));
const cursor = state.cursor;
state.cursor++;
if (cursor === self.#cachedMinCursor &&
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
state.detached = true;
self.#deleteConsumer(state);
return new IterResult(true, undefined);
}
// Check buffer limit. 'unbounded' and 'drop-newest' are rejected
// by shareSync().
if (self.#bufferedBytes >= self.#options.budget) {
switch (self.#options.backpressure) {
case 'strict': {
const error = new ERR_OUT_OF_RANGE(
'buffered bytes', `< ${self.#options.budget}`,
self.#bufferedBytes);
// Detach before throwing, as the async share does. Neither
// for...of nor a transform pipeline calls return() when
// next() throws, so a consumer left registered here would
// keep its cursor forever and wedge the other consumers.
state.detached = true;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
throw error;
}
case 'drop-oldest':
while (self.#bufferedBytes >= self.#options.budget &&
self.#buffer.length > 0) {
const evicted = self.#buffer.shift();
self.#bufferedBytes -= evicted.byteLength;
self.#bufferStart++;
}
for (const consumer of self.#consumers) {
if (consumer.cursor < self.#bufferStart) {
self.#deleteConsumerFromMin(consumer);
consumer.cursor = self.#bufferStart;
}
}
self.#recomputeMinCursor();
break;
}
}
self.#pullFromSource();
if (self.#sourceError !== kNoShareError) {
state.detached = true;
state.error = self.#sourceError;
self.#deleteConsumer(state);
throw state.error;
}
const newBufferIndex = state.cursor - self.#bufferStart;
if (newBufferIndex < self.#buffer.length) {
const chunk = self.#readEntry(self.#buffer.get(newBufferIndex));
const cursor = state.cursor;
state.cursor++;
if (cursor === self.#cachedMinCursor &&
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
state.detached = true;
self.#deleteConsumer(state);
return new IterResult(true, undefined);
}
return new IterResult(true, undefined);
},
return() {
state.detached = true;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
return new IterResult(true, undefined);
},
throw() {
state.detached = true;
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
return new IterResult(true, undefined);
},
}, null);
},
};
}
cancel(reason) {
if (this.#cancelled) return;
const hasReason = arguments.length > 0;
this.#cancelled = true;
if (hasReason) {
this.#sourceError = reason;
}
try {
const returnMethod = this.#sourceIterator?.return;
if (typeof returnMethod === 'function') {
FunctionPrototypeCall(returnMethod, this.#sourceIterator);
}
} catch {
// Cancellation has precedence over source cleanup errors.
}
for (const consumer of this.#consumers) {
if (hasReason) consumer.error = reason;
consumer.detached = true;
}
this.#consumers.clear();
}
[SymbolDispose]() {
this.cancel();
}
#pullFromSource() {
if (this.#sourceExhausted || this.#cancelled) return;
// A source that reads from its own share would re-enter its own
// iterator. Fail that read without touching the share's state; the
// source sees the error and may handle it.
if (this.#pulling) {
throw new ERR_INVALID_STATE(
'shareSync() source cannot be read while it is producing a value');
}
this.#pulling = true;
try {
this.#sourceIterator ||= this.#source[SymbolIterator]();
const result = this.#sourceIterator.next();
if (result.done) {
this.#sourceExhausted = true;
} else {
this.#bufferBatch(result.value);
}
} catch (error) {
this.#sourceError = error;
this.#sourceExhausted = true;
} finally {
this.#pulling = false;
}
}
#bufferBatch(batch) {
const entry = createBatchEntry(batch);
// 'drop-oldest' evicts whole entries. A single pulled batch can be much
// larger than the budget (for example when from() combines many values
// of a sync source), and evicting it would discard every chunk in it,
// including ones that no consumer has read. Split such batches so that
// eviction keeps the newest chunks that fit the budget.
const entries = this.#options.backpressure === 'drop-oldest' ?
splitBatchEntry(entry, this.#options.budget) : undefined;
if (entries === undefined) {
this.#buffer.push(entry);
} else {
for (let i = 0; i < entries.length; i++) {
this.#buffer.push(entries[i]);
}
}
this.#bufferedBytes += entry.byteLength;
}
#tryTrimBuffer() {
// Retain buffered data for consumers that attach while none are active.
// Without this, the last consumer detaching would discard entries that
// a late-joining consumer has not read and cannot get back from the
// source.
if (this.#consumers.size === 0) return;
if (this.#cachedMinCursorConsumers === 0) {
this.#recomputeMinCursor();
}
const trimCount = this.#cachedMinCursor - this.#bufferStart;
if (trimCount > 0) {
for (let i = 0; i < trimCount; i++) {
const evicted = this.#buffer.get(i);
this.#bufferedBytes -= evicted.byteLength;
}
this.#buffer.trimFront(trimCount);
this.#bufferStart = this.#cachedMinCursor;
}
}
#readEntry(entry) {
try {
return validateBatchEntry(entry);
} catch (error) {
this.cancel(error);
this.#buffer.clear();
this.#bufferedBytes = 0;
throw error;
}
}
#recomputeMinCursor() {
const { minCursor, minCursorConsumers } = getMinCursor(
this.#consumers, this.#bufferStart + this.#buffer.length);
this.#cachedMinCursor = minCursor;
this.#cachedMinCursorConsumers = minCursorConsumers;
}
#deleteConsumerFromMin(consumer) {
if (consumer.cursor === this.#cachedMinCursor) {
this.#cachedMinCursorConsumers--;
return this.#cachedMinCursorConsumers === 0;
}
return false;
}
#deleteConsumer(consumer) {
if (this.#consumers.delete(consumer)) {
return this.#deleteConsumerFromMin(consumer);
}
return false;
}
}
// =============================================================================
// Public API
// =============================================================================
function share(source, options = { __proto__: null }) {
// Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
const normalized = from(source);
options = converters.ShareOptions(options, {
__proto__: null,
context: 'options',
});
const {
budget = kMultiConsumerDefaultBudget,
backpressure = 'strict',
signal,
} = options;
validateBudget(budget);
const opts = {
__proto__: null,
budget,
backpressure,
signal,
};
const shareImpl = new ShareImpl(normalized, opts);
if (signal) {
shareImpl[kSetFactorySignal](signal);
}
return shareImpl;
}
function shareSync(source, options = { __proto__: null }) {
// Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
const normalized = fromSync(source);
options = converters.ShareSyncOptions(options, {
__proto__: null,
context: 'options',
});
const {
budget = kMultiConsumerDefaultBudget,
backpressure = 'strict',
} = options;
validateBudget(budget);
// A synchronous consumer can neither wait for the slowest consumer to
// release budget ('unbounded') nor keep pulling and discarding until it
// does ('drop-newest'): the slowest consumer cannot advance while the
// call is running.
if (backpressure === 'unbounded' || backpressure === 'drop-newest') {
throw new ERR_INVALID_ARG_VALUE(
'options.backpressure', backpressure,
`${backpressure} is not supported by shareSync()`);
}
const opts = {
__proto__: null,
budget,
backpressure,
};
return new SyncShareImpl(normalized, opts);
}
const Share = {
__proto__: null,
from(input, options) {
const protocol = getProtocolMethod(input, shareProtocol);
if (protocol !== undefined) {
const result = FunctionPrototypeCall(protocol, input, options);
if (result === null || typeof result !== 'object') {
throw new ERR_INVALID_RETURN_VALUE(
'an object', '[Symbol.for(\'Stream.shareProtocol\')]', result);
}
return result;
}
if (isAsyncIterable(input) || isSyncIterable(input)) {
return share(input, options);
}
throw new ERR_INVALID_ARG_TYPE(
'input', ['Shareable', 'AsyncIterable', 'Iterable'], input);
},
};
const SyncShare = {
__proto__: null,
fromSync(input, options) {
const protocol = getProtocolMethod(input, shareSyncProtocol);
if (protocol !== undefined) {
const result = FunctionPrototypeCall(protocol, input, options);
if (result === null || typeof result !== 'object') {
throw new ERR_INVALID_RETURN_VALUE(
'an object', '[Symbol.for(\'Stream.shareSyncProtocol\')]', result);
}
return result;
}
if (isSyncIterable(input)) {
return shareSync(input, options);
}
throw new ERR_INVALID_ARG_TYPE(
'input', ['SyncShareable', 'Iterable'], input);
},
};
module.exports = {
Share,
SyncShare,
share,
shareSync,
};