import { identity } from '../util/identity.js'; import { wrapWithAbort } from './operators/withabort.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>; } /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @returns {(Promise<[T, T2] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( source: AsyncIterable, source2: AsyncIterable ): Promise<[T, T2] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @returns {(Promise<[T, T2, T3] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable ): Promise<[T, T2, T3] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @param {AsyncIterable} source4 Fourth async-iterable source. * @returns {(Promise<[T, T2, T3, T4] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable ): Promise<[T, T2, T3, T4] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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} source First async-iterable source. * @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 {(Promise<[T, T2, T3, T4, T5] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable ): Promise<[T, T2, T3, T4, T5] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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} source First async-iterable source. * @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 {(Promise<[T, T2, T3, T4, T5, T6] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable, source6: AsyncIterable ): Promise<[T, T2, T3, T4, T5, T6] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @template T The type of the elements in the first source sequence. * @param {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @returns {(Promise<[T] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable ): Promise<[T] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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 {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @returns {(Promise<[T, T2] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable, source2: AsyncIterable ): Promise<[T, T2] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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 {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @returns {(Promise<[T, T2, T3] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable ): Promise<[T, T2, T3] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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 {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @param {AsyncIterable} source2 Second async-iterable source. * @param {AsyncIterable} source3 Third async-iterable source. * @param {AsyncIterable} source4 Fourth async-iterable source. * @returns {(Promise<[T, T2, T3, T4] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable ): Promise<[T, T2, T3, T4] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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 {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @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 {(Promise<[T, T2, T3, T4, T5] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable ): Promise<[T, T2, T3, T4, T5] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @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 {AbortSignal} signal An abort signal used for cancellation at any time. * @param {AsyncIterable} source First async-iterable source. * @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 {(Promise<[T, T2, T3, T4, T5, T6] | undefined>)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, source: AsyncIterable, source2: AsyncIterable, source3: AsyncIterable, source4: AsyncIterable, source5: AsyncIterable, source6: AsyncIterable ): Promise<[T, T2, T3, T4, T5, T6] | undefined>; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @template T The type of the elements in the source sequences. * @param {...AsyncIterable[]} sources The source sequences. * @returns {(Promise)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin(...sources: AsyncIterable[]): Promise; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @template T The type of the elements in the source sequences. * @param {AbortSignal} signal An abort signal used for cancellation at any time. * @param {...AsyncIterable[]} sources The source sequences. * @returns {(Promise)} An async-iterable sequence with an array of all the last elements of all sequences. */ export function forkJoin( signal: AbortSignal, ...sources: AsyncIterable[] ): Promise; /** * Runs all specified async-iterable sequences in parallel and collects their last elements. * * @template T The type of the elements in the source sequences. * @param {...any[]} sources Async-iterable sequence to collect the last elements for. * @returns {(Promise)} An async-iterable sequence with an array of all the last elements of all sequences. */ export async function forkJoin(...sources: any[]): Promise { let signal = sources.shift() as AbortSignal | undefined; if (!(signal instanceof AbortSignal)) { sources.unshift(signal); signal = undefined; } const length = sources.length; const iterators = new Array>(length); const nexts = new Array>>>(length); let active = length; const values = new Array(length); const hasValues = new Array(length); hasValues.fill(false); for (let i = 0; i < length; i++) { const iterator = wrapWithAbort(sources[i], signal)[Symbol.asyncIterator](); iterators[i] = iterator; nexts[i] = wrapPromiseWithIndex(iterator.next(), i); } while (active > 0) { const next = safeRace(nexts); const { value: next$, index } = await next; if (next$.done) { nexts[index] = >>>NEVER_PROMISE; active--; } else { const iterator$ = iterators[index]; nexts[index] = wrapPromiseWithIndex(iterator$.next(), index); hasValues[index] = true; values[index] = next$.value; } } if (hasValues.length > 0 && hasValues.every(identity)) { return values; } return undefined; }