Handle Concurrency

concurrent is a function that can handle multiple asynchronous values at once.

In javascript, there is a function to evaluate multiple promise values at the same time with Promise.all. However, it cannot handle the load of concurrent requests and cannot handle requests for infinite enumerable data sets. concurrent can handle asynchronous requests of infinite datasets and can control the request size of the load.

// prettier-ignore
import { pipe, toAsync, range, map, filter, take, each, concurrent } from "@fxts/core";

const fetchApi = (page) =>
  new Promise((resolve) => setTimeout(() => resolve(page), 1000));

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi), // 0,1,2,3,4,5
  filter((a) => a % 2 === 0),
  take(3), // 0,2,4
  concurrent(3), // If this line does not exist, it will take a total of 6 seconds.
  each(console.log), // 2 seconds
);

You can see that it takes 6 seconds when requesting one by one but takes 2 seconds when requesting using concurrent

Useful Example

A more practical code is below.

If the position of concurrent in the code above is as follows, would the result be different? No, It would be same! Note that concurrent always applies to Iterable before the length is changed.

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi),
  concurrent(3),
  filter((a) => a % 2 === 0),
  take(3),
  each(console.log),
);

If you want to sequentially evaluate up to map one by one, and evaluate the asynchronous predicate of filter three at the same time, you should write the code below:

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi),
  toArray,
  filter((a) => delay(100, a % 2 === 0)),
  take(3),
  concurrent(3),
  each(console.log),
);

Window or pool: `concurrent` vs `concurrentPool`

concurrent(n) evaluates in fixed windows: it fires n requests, waits for the window to settle, then starts the next one — ideal for batch-style workloads of similar size.

When task durations vary, a window waits for its slowest member. concurrentPool(n) keeps a rolling pool instead — the moment one task settles, the next begins — while still yielding results in input order:

await pipe(
  toAsync(urls),
  map(fetchItem),
  concurrentPool(5), // at most 5 in flight, refilled as each settles
  toArray,
);

Handle Concurrency ​

Useful Example ​

Window or pool: concurrent vs concurrentPool ​

Handle Concurrency

Useful Example

Window or pool: `concurrent` vs `concurrentPool`