Function asyncIteratorMap

client/js/src/internal/asyncIteratorMap.ts:21–72

(
    inputIterator: AsyncGenerator<I>,
    mapper: (item: I) => Promise<O>,
    concurrency: number = DEFAULT_CONCURRENCY,
    signal?: AbortSignal,
)

 * @param concurrency - The concurrency limit. How many parallel async mapper calls are allowed.
 * @returns An async iterator that yields the mapped values.
 */
export async function* asyncIteratorMap<I, O>(
    inputIterator: AsyncGenerator<I>,
    mapper: (item: I) => Promise<O>,
    concurrency: number = DEFAULT_CONCURRENCY,
    signal?: AbortSignal,
): AsyncGenerator<O> {
    let done = false;
    let hasInputError = false;
    let inputError: unknown | undefined;

    const executing = new Set<Promise<void>>();
    const results: Array<Promise<O>> = [];

    const pump = async () => {
        let next;
        try {
            next = await inputIterator.next();
        } catch (error) {
            hasInputError = true;
            inputError = error;
            done = true;
            return;
        }

        if (next.done) {
            done = true;
            return;
        }

        const promise = mapper(next.value)
            .then((result) => {
                results.push(Promise.resolve(result));
            })
            .catch((error) => {
                results.push(Promise.reject(error));
            });
        executing.add(promise);
        void promise.finally(() => executing.delete(promise));
    };

    while (!done || executing.size > 0 || results.length > 0) {
        if (signal?.aborted) {
            throw new AbortError(c('Error').t`Operation aborted`);
        }
        while (!done && executing.size < concurrency) {
            await pump();
        }

        if (results.length > 0) {
            yield await results.shift()!;
        } else if (executing.size > 0) {
            // Wait for at least one task to complete
            await Promise.race(Array.from(executing));
        }