mapConcurrent

mapConcurrent<T, R>(source, concurrency, mapper): AsyncGenerator<R>

Defined in: packages/simply-core/src/async/mapConcurrent.ts:61

Map over source with at most concurrency calls to mapper in flight at once, starting the next item the instant a slot frees up rather than waiting for sibling items pulled at the same time to finish too.

This is the streaming counterpart to mapChunked: mapChunked requires the full input as an in-memory array and advances in fixed-size, wait-for-the-whole-batch steps, which is right for bounding concurrency against an already-loaded list. mapConcurrent instead consumes source lazily — one item at a time, only as pool slots come free — so a huge source (e.g. rows streamed from a large CSV) never has to be fully buffered in memory, and a pool slot doesn’t sit idle waiting on the slowest item in an arbitrary batch.

Results are yielded in completion order, not source order — whichever mapper call finishes first is yielded first. Callers that need source order should use mapChunked (or sort the results themselves, if T/R carry an index).

mapper is expected not to reject: this stays a plain concurrency pool, not an error-handling layer, so a rejection propagates out of this generator immediately (ending iteration) rather than being caught, retried, or turned into a result. A caller that wants a run to continue past individual failures should have mapper catch its own errors and resolve with a result that represents them.

Type Parameters

T

T

R

R

Parameters

source

AsyncIterable<T, any, any> | Iterable<T, any, any>

The items to map over, pulled lazily.

concurrency

number

Maximum number of mapper calls in flight at once. Must be at least 1.

mapper

(item) => Promise<R>

Called once per item, as soon as a pool slot is free for it.

Returns

AsyncGenerator<R>

Yields

Each mapper result, in completion order.

Throws

If concurrency is less than 1.