Creates a task interleaving several runs of another task over the same feed.
Runs draw from the same source, each item going to exactly one of them, processed through that run's own invocation
of task and emitted as soon as it is ready. Runs are interleaved on the event loop rather than executed in
parallel, so running a task that blocks the event loop concurrently buys nothing.
concurrency caps the number of items being processed at the same time, which is also how far ahead of the
downstream consumer items are pulled from the source: raising it trades memory for throughput. All runs start
together when the feed is opened, so task is invoked exactly concurrency times whether or not the source is
fast enough to keep every run busy.
Warning
Output order is not preserved: results are emitted as each run completes, so they interleave and overtake each
other according to how long every item takes to process.
Caution
concurrency is not a rate limit: nothing paces the runs, so a source that never runs dry keeps concurrency
items permanently in flight. Pace the work inside task instead, for instance with a
Throttle from the @metreeca/core/async
module.
Caution
Run stateless tasks only. task is invoked once per run, so state it keeps per invocation becomes per-run
rather than feed-wide: distinct deduplicates within a run, take yields its quota to each one.
State captured in its closure is shared instead, and hit concurrently.
Note
When an error occurs, all pending operations are awaited (but not failed) before the error is thrown, to prevent
resource leaks. Runs failing while the consumer is idle report their error when the feed is next advanced,
rather than escaping as unhandled rejections.
Note
Closing the feed early waits for the in-flight pulls and the running tasks to settle before the source is
closed, so a source idling between items delays it; failures reported while closing are suppressed if an error
is already propagating.
Type Parameters
V
The type of input items
R
The type of output items
Parameters
concurrency: number
The number of concurrent runs; values less than 1 are treated as 1, that is, as sequential
processing
Creates a task interleaving several runs of another task over the same feed.
Runs draw from the same source, each item going to exactly one of them, processed through that run's own invocation of
taskand emitted as soon as it is ready. Runs are interleaved on the event loop rather than executed in parallel, so running a task that blocks the event loop concurrently buys nothing.concurrencycaps the number of items being processed at the same time, which is also how far ahead of the downstream consumer items are pulled from the source: raising it trades memory for throughput. All runs start together when the feed is opened, sotaskis invoked exactlyconcurrencytimes whether or not the source is fast enough to keep every run busy.Output order is not preserved: results are emitted as each run completes, so they interleave and overtake each other according to how long every item takes to process.
concurrencyis not a rate limit: nothing paces the runs, so a source that never runs dry keepsconcurrencyitems permanently in flight. Pace the work insidetaskinstead, for instance with a Throttle from the@metreeca/core/asyncmodule.Run stateless tasks only.
taskis invoked once per run, so state it keeps per invocation becomes per-run rather than feed-wide: distinct deduplicates within a run, take yields its quota to each one. State captured in its closure is shared instead, and hit concurrently.When an error occurs, all pending operations are awaited (but not failed) before the error is thrown, to prevent resource leaks. Runs failing while the consumer is idle report their error when the feed is next advanced, rather than escaping as unhandled rejections.
Closing the feed early waits for the in-flight pulls and the running tasks to settle before the source is closed, so a source idling between items delays it; failures reported while closing are suppressed if an error is already propagating.