Usage

// Process images concurrently, resizing each image to a standard size.
const images = await parallel(2, imageFiles, async (file) => {
  return await resizeImage(file);
});

Installation

CLI

npx atmx add helper parallel

Manually

Copy and paste the following method into @/utils/helpers/parallel.ts:

import { fork } from "@/helpers/fork.ts";
import { list } from "@/helpers/list.ts";
import { sort } from "@/helpers/sort.ts";
import { tryit } from "@/helpers/tryit.ts";

type WorkItemResult<K> = {
index: number;
result: K;
error: any;
};

/**
* Executes many async functions in parallel. Returns the results from
* all functions as an array. After all functions have resolved, if
* any errors were thrown, they are rethrown in an instance of
* AggregateError.
*
* @example
* // Process images concurrently, resizing each image to a standard size.
* const images = await parallel(2, imageFiles, async (file) => {
*   return await resizeImage(file)
* })
*/
export async function parallel<T, K>(
limit: number,
array: readonly T[],
func: (item: T) => Promise<K>,
): Promise<K[]> {
const work = array.map((item, index) => ({
index,
item,
}));
// Process array items
const processor = async (res: (value: WorkItemResult<K>[]) => void) => {
const results: WorkItemResult<K>[] = [];
while (true) {
  const next = work.pop();
  if (!next) {
    return res(results);
  }
  const [error, result] = await tryit(func)(next.item);
  results.push({
    error,
    result: result as K,
    index: next.index,
  });
}
};
// Create queues
const queues = list(1, limit).map(() => new Promise(processor));
// Wait for all queues to complete
const itemResults = (await Promise.all(queues)) as WorkItemResult<K>[][];
const [errors, results] = fork(
sort(itemResults.flat(), (r) => r.index),
(x) => !!x.error,
);
if (errors.length > 0) {
throw new AggregateError(errors.map((error) => error.error));
}
return results.map((r) => r.result);
}