import { IterableX } from '../iterablex.js'; import { RefCountList } from './_refcountlist.js'; import { create } from '../create.js'; import { OperatorFunction } from '../../interfaces.js'; class PublishedBuffer extends IterableX { private _buffer: RefCountList; private _source: Iterator; private _error: any; private _stopped = false; constructor(source: Iterator) { super(); this._source = source; this._buffer = new RefCountList(0); } // eslint-disable-next-line complexity private *_getIterable(i: number): Iterable { try { while (1) { let hasValue = false; let current = {}; if (i >= this._buffer.count) { if (!this._stopped) { try { const next = this._source.next(); hasValue = !next.done; // eslint-disable-next-line max-depth if (hasValue) { current = next.value; } } catch (e) { this._error = e; this._stopped = true; } } if (this._stopped) { if (this._error) { throw this._error; } else { break; } } if (hasValue) { this._buffer.push(current); } } else { hasValue = true; } if (hasValue) { yield this._buffer.get(i); } else { break; } // eslint-disable-next-line no-param-reassign i++; } } finally { this._buffer.done(); } } [Symbol.iterator](): Iterator { this._buffer.readerCount++; return this._getIterable(this._buffer.count)[Symbol.iterator](); } } /** * Creates a buffer with a view over the source sequence, causing each iterator to obtain access to the * remainder of the sequence from the current index in the buffer. * * @template TSource Source sequence element type. * @returns {OperatorFunction} Buffer enabling each iterator to retrieve elements from * the shared source sequence, starting from the index at the point of obtaining the enumerator. */ export function publish(): OperatorFunction; /** * Buffer enabling each iterator to retrieve elements from the shared source sequence, starting from the * index at the point of obtaining the iterator. * * @template TSource Source sequence element type. * @template TResult Result sequence element type. * @param {(value: Iterable) => Iterable} [selector] Selector function with published * access to the source sequence for each iterator. * @returns {OperatorFunction} Sequence resulting from applying the selector function to the * published view over the source sequence. */ export function publish( selector?: (value: Iterable) => Iterable ): OperatorFunction; /** * Buffer enabling each iterator to retrieve elements from the shared source sequence, starting from the * index at the point of obtaining the iterator. * * @template TSource Source sequence element type. * @template TResult Result sequence element type. * @param {(value: Iterable) => Iterable} [selector] Selector function with published * access to the source sequence for each iterator. * @returns {(OperatorFunction)} Sequence resulting from applying the selector function to the * published view over the source sequence. */ export function publish( selector?: (value: Iterable) => Iterable ): OperatorFunction { return function publishOperatorFunction(source: Iterable): IterableX { return selector ? create(() => selector(publish()(source))[Symbol.iterator]()) : new PublishedBuffer(source[Symbol.iterator]()); }; }