import { AsyncIterableX } from '../asynciterablex.js'; import { OperatorAsyncFunction } from '../../interfaces.js'; import { wrapWithAbort } from './withabort.js'; import { throwIfAborted } from '../../aborterror.js'; import { identity } from '../../util/identity.js'; import { safeRace } from '../../util/safeRace.js'; // eslint-disable-next-line @typescript-eslint/no-empty-function const NEVER_PROMISE = new Promise(() => {}); type MergeResult = { value: T; index: number }; function wrapPromiseWithIndex(promise: Promise, index: number) { return promise.then((value) => ({ value, index })) as Promise>; } /** @ignore */ export class WithLatestFromAsyncIterable extends AsyncIterableX { private _source: AsyncIterable; private _others: AsyncIterable[]; constructor(source: AsyncIterable, others: AsyncIterable[]) { super(); this._source = source; this._others = others; } async *[Symbol.asyncIterator](signal?: AbortSignal) { throwIfAborted(signal); const length = this._others.length; const newLength = length + 1; const iterators = new Array>(newLength); const nexts = new Array>>>(newLength); let hasValueAll = false; const hasValue = new Array(length); const values = new Array(length); hasValue.fill(false); for (let i = 0; i < length; i++) { const iterator = wrapWithAbort(this._others[i], signal)[Symbol.asyncIterator](); iterators[i] = iterator; nexts[i] = wrapPromiseWithIndex(iterator.next(), i); } const it = wrapWithAbort(this._source, signal)[Symbol.asyncIterator](); iterators[length] = it; nexts[length] = wrapPromiseWithIndex(it.next(), length); for (;;) { const next = safeRace(nexts); const { value: { value: value$, done: done$ }, index, } = await next; if (index === length) { if (done$) { break; } const iterator$ = iterators[index]; nexts[index] = wrapPromiseWithIndex(iterator$.next(), index); if (hasValueAll) { yield [value$, ...values]; } } else if (done$) { nexts[index] = >>>NEVER_PROMISE; } else { values[index] = value$; hasValue[index] = true; hasValueAll = hasValue.every(identity); const iterator$ = iterators[index]; nexts[index] = wrapPromiseWithIndex(iterator$.next(), index); } } } } /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the first source sequence. * @template T2 The type of the elements in the second source sequence. * @param {AsyncIterable} source2 Second async-iterable source. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom( source2: AsyncIterable ): OperatorAsyncFunction; /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the first source sequence. * @template T2 The type of the elements in the second source sequence. * @template T3 The type of the elements in the third source sequence. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom( source2: AsyncIterable, source3: AsyncIterable ): OperatorAsyncFunction; /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the first source sequence. * @template T2 The type of the elements in the second source sequence. * @template T3 The type of the elements in the third source sequence. * @template T4 The type of the elements in the fourth source sequence. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @param {AsyncIterable} source4 Fourth async-iterable source. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom( source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable ): OperatorAsyncFunction; /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the first source sequence. * @template T2 The type of the elements in the second source sequence. * @template T3 The type of the elements in the third source sequence. * @template T4 The type of the elements in the fourth source sequence. * @template T5 The type of the elements in the fifth source sequence. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @param {AsyncIterable} source4 Fourth async-iterable source. * @param {AsyncIterable} source5 Fifth async-iterable source. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom( source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable ): OperatorAsyncFunction; /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the first source sequence. * @template T2 The type of the elements in the second source sequence. * @template T3 The type of the elements in the third source sequence. * @template T4 The type of the elements in the fourth source sequence. * @template T5 The type of the elements in the fifth source sequence. * @template T6 The type of the elements in the sixth source sequence. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @param {AsyncIterable} source4 Fourth async-iterable source. * @param {AsyncIterable} source5 Fifth async-iterable source. * @param {AsyncIterable} source6 Sixth async-iterable source. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom( source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable, source6: AsyncIterable ): OperatorAsyncFunction; /** * Merges multiple async-iterable sequences into one async-iterable sequence by combining each element * from the first source with the latest element from the other sources, if any. * * @template T The type of the elements in the source sequences. * @param {...AsyncIterable[]} sources The source sequences. * @returns {OperatorAsyncFunction} An async-iterable sequence containing the result of combining * each element of the first source with the latest element from the second source, if any as an array. */ export function withLatestFrom(...sources: AsyncIterable[]): OperatorAsyncFunction; export function withLatestFrom(...sources: any[]): OperatorAsyncFunction { return function withLatestFromOperatorFunction(source: AsyncIterable) { return new WithLatestFromAsyncIterable(source, sources); }; }