`concurrentPool`

Like concurrent, but yields results in completion order — whichever finishes first comes out first.

FxAsyncIterable<A> concurrentPoolAsync<A>(int length, FxAsyncIterable<A> iterable) FxAsync<T> FxAsync.concurrentPool(int length) // chain

Lecture

concurrentPool(n) keeps up to n requests in flight against the upstream source, exactly like concurrent(n) — but it hands results back to you in the order they complete, not the order they started in. Think of it as a worker pool: as soon as one slot frees up, the next pending item is launched into it, and whichever finishes first is yielded first. This matches FxTS's concurrentPool and is the right tool when you don't care which result came from which input — you just want to react to results as soon as each one is ready (e.g. updating a progress list as each fetch lands), rather than blocking on the slowest one to preserve order.

Unlike concurrent, which is driven by the downstream demand marker, concurrentPool eagerly keeps its pool full: from the first pull onward it holds up to n requests in flight no matter how many consumers are waiting. Even a one-pull-at-a-time terminal like .toList() or .each() gets the full overlap — and sees results in the order they finish.

Demo 1 · Completion order

Item 1 is slowest (300ms) and item 2 is fastest (100ms) — the result comes out fastest-first, straight from .toList():

Demo 2 · Contrast with concurrent

Same delays, same pool size of 3 — the only difference is which order the results land in:

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  final delays = [300, 100, 200]; // item1 is slowest, item2 is fastest

Future<FxAsync<String>> makeChain() async => fx([1, 2, 3])
      .toAsync()
      .map((i) => delay(Duration(milliseconds: delays[i - 1]), 'item$i'));

// concurrent(3): still 3 in flight at once, but the *result* order always
  // matches the source order.
  final orderedIt = (await makeChain()).concurrent(3).iterator;
  final ordered =
      (await Future.wait([orderedIt.next(), orderedIt.next(), orderedIt.next()]))
          .map((r) => r.value)
          .toList();
  print(ordered); // [item1, item2, item3] — source order

// concurrentPool(3): same 3-in-flight budget, but the *result* order
  // matches whichever finishes first.
  final poolIt = (await makeChain()).concurrentPool(3).iterator;
  final pooled =
      (await Future.wait([poolIt.next(), poolIt.next(), poolIt.next()]))
          .map((r) => r.value)
          .toList();
  print(pooled); // [item2, item3, item1] — completion order
}

Try it yourself

Exercise: this pool size of 1 processes items one at a time, in launch order. Bump it to 3 so all three race at once, and watch the printed order switch to completion order.

Related: concurrent — order-preserving variant · toAsync — the pull-based model this relies on · Stream bridges — apply concurrentPool before toStream() · debounce — rate-limiting for callbacks