Function accumulator

Extension/src/Utility/Async/iterators.ts:77–135

(...iterables: Some<T>[])

};

export function accumulator<T>(...iterables: Some<T>[]): AsynchIterable<T> {
    const iterators = new Map<number, Promise<Cursor<T>>>();
    let completeWhenEmpty = iterables.length > 0; // if we are given any items, they we auto-complete when we run out (so an add after the last item is yielded will throw)
    const signal = new Signal<boolean>();

    const result = combiner(iterables) as unknown as AsynchIterable<T>;
    result.add = (...iterables: (Some<T> | undefined | Promise<undefined | T>)[]) => {
        for (const iterable of iterables) {
            iterators.set(iterators.size + 1, awaitNext({ identity: iterators.size + 1, iterator: asyncOf(iterable)[Symbol.asyncIterator]() }));
        }
        signal.resolve(true);
    };

    result.autoComplete = (shouldAutocomplete: boolean) => { completeWhenEmpty = shouldAutocomplete; return result; };
    result.complete = () => signal.dispose(completeWhenEmpty = true);
    result.reiterable = () => reiterable(result);

    return result;

    async function awaitNext(element: Cursor<T>): Promise<Cursor<T>> {
        element.result = undefined; // drop the previous result before awaiting the next
        element.result = await element.iterator.next();
        return element;
    }

    async function* combiner(iterables: Some<T>[]) {
        iterables.forEach((iterable, index) => iterators.set(index, awaitNext({ identity: index, iterator: asyncOf(iterable)[Symbol.asyncIterator]() })));

        // Loop until all iterators are done, removing them as they complete
        do {
            if (!iterators.size && !completeWhenEmpty) {
                await signal; // wait for a new item to be added, or for the complete signal
            }

            while (iterators.size) {
                const element = await race([...iterators.values()]);

                // Is that iterator done?
                if (element.result!.done) {
                    iterators.delete(element.identity);
                    continue;
                }

                // Yield the result from the iterator, and await the next item
                const { value } = element.result!;
                iterators.set(element.identity, awaitNext(element));
                if (value !== undefined && value !== null) {
                    yield value;
                }

            }

        } while (!completeWhenEmpty);
        signal.dispose(false);

        // prevent any more iterators from being added
        result.add = () => { throw new Error('AsyncIterable is finished'); };
    }