/**
 * @packageDocumentation
 *
 * Takes an (async) iterable that emits promise-returning functions, invokes them in parallel up to the concurrency limit and emits the results as they become available, optionally in the same order as the input
 *
 * @example
 *
 * ```javascript
 * import parallel from 'it-parallel'
 * import all from 'it-all'
 * import delay from 'delay'
 *
 * // This can also be an iterator, async iterator, generator, etc
 * const input = [
 *   async () => {
 *     console.info('start 1')
 *     await delay(500)
 *
 *     console.info('end 1')
 *     return 1
 *   },
 *   async () => {
 *     console.info('start 2')
 *     await delay(200)
 *
 *     console.info('end 2')
 *     return 2
 *   },
 *   async () => {
 *     console.info('start 3')
 *     await delay(100)
 *
 *     console.info('end 3')
 *     return 3
 *   }
 * ]
 *
 * const result = await all(parallel(input, {
 *   concurrency: 2
 * }))
 *
 * // output:
 * // start 1
 * // start 2
 * // end 2
 * // start 3
 * // end 3
 * // end 1
 *
 * console.info(result) // [2, 3, 1]
 * ```
 *
 * If order is important, pass `ordered: true` as an option:
 *
 * ```javascript
 * const result = await all(parallel(input, {
 *   concurrency: 2,
 *   ordered: true
 * }))
 *
 * // output:
 * // start 1
 * // start 2
 * // end 2
 * // start 3
 * // end 3
 * // end 1
 *
 * console.info(result) // [1, 2, 3]
 * ```
 */
import defer from 'p-defer';
const CustomEvent = globalThis.CustomEvent ?? Event;
/**
 * Takes an (async) iterator that emits promise-returning functions,
 * invokes them in parallel and emits the results as they become available but
 * in the same order as the input
 */
export default async function* parallel(source, options = {}) {
    let concurrency = options.concurrency ?? Infinity;
    if (concurrency < 1) {
        concurrency = Infinity;
    }
    const ordered = options.ordered ?? false;
    const emitter = new EventTarget();
    const ops = [];
    let slotAvailable = defer();
    let resultAvailable = defer();
    let sourceFinished = false;
    let sourceErr;
    let opErred = false;
    let returned = false;
    emitter.addEventListener('task-complete', () => {
        resultAvailable.resolve();
    });
    void Promise.resolve().then(async () => {
        try {
            for await (const task of source) {
                if (ops.length === concurrency) {
                    slotAvailable = defer();
                    await slotAvailable.promise;
                }
                if (opErred || returned) {
                    break;
                }
                const op = {
                    done: false
                };
                ops.push(op);
                task()
                    .then(result => {
                    op.done = true;
                    op.ok = true;
                    op.value = result;
                    emitter.dispatchEvent(new CustomEvent('task-complete'));
                }, err => {
                    op.done = true;
                    op.err = err;
                    emitter.dispatchEvent(new CustomEvent('task-complete'));
                });
            }
            sourceFinished = true;
            emitter.dispatchEvent(new CustomEvent('task-complete'));
        }
        catch (err) {
            sourceErr = err;
            emitter.dispatchEvent(new CustomEvent('task-complete'));
        }
    });
    function valuesAvailable() {
        if (ordered) {
            return ops[0]?.done;
        }
        return Boolean(ops.find(op => op.done));
    }
    function* yieldOrderedValues() {
        while ((ops.length > 0) && ops[0].done) {
            const op = ops[0];
            ops.shift();
            if (op.ok) {
                yield op.value;
            }
            else {
                // allow the source to exit
                opErred = true;
                slotAvailable.resolve();
                throw op.err;
            }
            slotAvailable.resolve();
        }
    }
    function* yieldUnOrderedValues() {
        // more values can become available while we wait for `yield`
        // to return control to this function
        while (valuesAvailable()) {
            for (let i = 0; i < ops.length; i++) {
                if (ops[i].done) {
                    const op = ops[i];
                    ops.splice(i, 1);
                    i--;
                    if (op.ok) {
                        yield op.value;
                    }
                    else {
                        opErred = true;
                        slotAvailable.resolve();
                        throw op.err;
                    }
                    slotAvailable.resolve();
                }
            }
        }
    }
    try {
        while (true) {
            if (!valuesAvailable()) {
                resultAvailable = defer();
                await resultAvailable.promise;
            }
            if (sourceErr != null) {
                // the source threw an error, propagate it
                throw sourceErr;
            }
            if (ordered) {
                yield* yieldOrderedValues();
            }
            else {
                yield* yieldUnOrderedValues();
            }
            if (sourceErr != null) {
                // if the source yields an array that is `yield *`, it can throw while the
                // onward consumer is processing the array contents - make sure we
                // propagate the error
                // eslint-disable-next-line @typescript-eslint/only-throw-error
                throw sourceErr;
            }
            if (sourceFinished && ops.length === 0) {
                // not waiting for any results and no more tasks so we are done
                break;
            }
        }
    }
    finally {
        // the consumer has stopped iterating before the source was exhausted -
        // let the source reader exit so the source iterator is returned and any
        // resources it holds are freed
        returned = true;
        slotAvailable.resolve();
    }
}
//# sourceMappingURL=index.js.map