Async Map With a Concurrency Limit

export async function mapWithConcurrency(items, limit, mapper, options = {}) {
  if (!Array.isArray(items)) {
    throw new TypeError("items must be an array");
  }
  if (!Number.isInteger(limit) || limit < 1) {
    throw new RangeError("limit must be a positive integer");
  }
  if (typeof mapper !== "function") {
    throw new TypeError("mapper must be a function");
  }
  const { signal, stopOnError = true } = options;
  if (
    signal !== undefined &&
    (typeof signal.aborted !== "boolean" || typeof signal.addEventListener !== "function")
  ) {
    throw new TypeError("signal must be an AbortSignal");
  }

  const results = new Array(items.length);
  let nextIndex = 0;
  let failed = false;
  let firstError;

  const worker = async () => {
    while (true) {
      if (signal?.aborted) {
        throw signal.reason ?? new DOMException("Aborted", "AbortError");
      }
      if (failed && stopOnError) return;
      const index = nextIndex;
      nextIndex += 1;
      if (index >= items.length) return;
      try {
        results[index] = await mapper(items[index], index);
      } catch (error) {
        if (!failed) {
          failed = true;
          firstError = error;
        }
        if (stopOnError) return;
      }
    }
  };

  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, () => worker()),
  );

  if (failed && stopOnError) throw firstError;
  return results;
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const order = [];
const squares = await mapWithConcurrency([1, 2, 3, 4, 5], 2, async (value) => {
  order.push(`start ${value}`);
  await sleep(value * 5);
  order.push(`end ${value}`);
  return value ** 2;
});
console.log(squares);
console.log(order.join(" | "));

const failure = await mapWithConcurrency([1, 2, 3], 2, async (value) => {
  if (value === 2) throw new Error(`failed on ${value}`);
  return value;
}).catch((error) => error.message);
console.log(failure);