import { AsyncIterableX } from '../asynciterablex.js'; import { IRefCountList, MaxRefCountList, RefCountList, } from '../../iterable/operators/_refcountlist.js'; import { create } from '../create.js'; import { OperatorAsyncFunction } from '../../interfaces.js'; import { throwIfAborted } from '../../aborterror.js'; /** @ignore */ export class MemoizeAsyncBuffer extends AsyncIterableX { protected _source: AsyncIterator; protected _buffer: IRefCountList; protected _shared: Promise> | null; protected _error: any; protected _stopped: boolean; constructor(source: AsyncIterator, buffer: IRefCountList) { super(); this._error = null; this._shared = null; this._stopped = false; this._source = source; this._buffer = buffer; } [Symbol.asyncIterator](signal?: AbortSignal) { throwIfAborted(signal); return this._getIterable(0); } protected async *_getIterable(offset = 0) { let i = offset - 1; let done: boolean | undefined = false; const buffer = this._buffer; try { do { if (++i < buffer.count) { yield buffer.get(i); continue; } if (this._stopped) { throw this._error; } if (this._shared === null) { this._shared = this._source.next().then((r) => { this._shared = null; if (!r.done) { buffer.push(r.value); } return r; }); } ({ done } = await this._shared.catch((e) => { this._error = e; this._stopped = true; throw e; })); if (!done) { yield buffer.get(i); } } while (!done); } finally { buffer.done(); } } } /** * Creates a buffer with a view over the source sequence, causing a specified number of iterators to obtain access * to all of the sequence's elements without causing multiple enumerations over the source. * @template TSource Source sequence element type. * @param {number} [readerCount] Number of iterators that can access the underlying buffer. * Once every iterator has obtained an element from the buffer, the element is removed from the buffer. * @returns {OperatorAsyncFunction} Buffer enabling a specified number of iterators to retrieve all * elements from the shared source sequence, without duplicating source iteration side-effects. */ export function memoize(readerCount?: number): OperatorAsyncFunction; /** * Memoizes the source sequence within a selector function where a specified number of iterators can get access * to all of the sequence's elements without causing multiple iterations over the source. * * @template TSource Source sequence element type. * @template TResult Result sequence element type. * @param {number} [readerCount] Number of iterators that can access the underlying buffer. Once every * iterator has obtained an element from the buffer, the element is removed from the buffer. * @param {(value: AsyncIterableX) => AsyncIterable} [selector] Selector function with memoized access * to the source sequence for a specified number of iterators. * @returns {OperatorAsyncFunction} Sequence resulting from applying the selector function to the * memoized view over the source sequence. */ export function memoize( readerCount?: number, selector?: (value: AsyncIterable) => AsyncIterable ): OperatorAsyncFunction; /** * Memoizes the source sequence within a selector function where a specified number of iterators can get access * to all of the sequence's elements without causing multiple iterations over the source. * * @template TSource Source sequence element type. * @template TResult Result sequence element type. * @param {number} [readerCount=-1] Number of iterators that can access the underlying buffer. Once every * iterator has obtained an element from the buffer, the element is removed from the buffer. * @param {(value: AsyncIterable) => AsyncIterable} [selector] Selector function with memoized access * to the source sequence for a specified number of iterators. * @returns {(OperatorAsyncFunction)} Sequence resulting from applying the selector function to the * memoized view over the source sequence. */ export function memoize( readerCount = -1, selector?: (value: AsyncIterable) => AsyncIterable ): OperatorAsyncFunction { return function memoizeOperatorFunction( source: AsyncIterable ): AsyncIterableX { if (!selector) { return readerCount === -1 ? new MemoizeAsyncBuffer( source[Symbol.asyncIterator](), new MaxRefCountList() ) : new MemoizeAsyncBuffer( source[Symbol.asyncIterator](), new RefCountList(readerCount) ); } return create(() => selector!(memoize(readerCount)(source))[Symbol.asyncIterator]() ); }; }