runWithConcurrency<T> function

Future<void> runWithConcurrency<T>(
  1. Iterable<T> items,
  2. int concurrency,
  3. Future<void> action(
    1. T item
    )
)

Run action over items with at most concurrency in flight at once. Mirrors internxt-dart's runWithConcurrency: a ChunkSemaphore gates how many item futures are active; the rest queue. This is the file-level (Step 2) batch primitive — each whole-file transfer is one item, and the per-file chunk concurrency (Step 1) composes underneath. Completes once every item has finished. If action throws for an item, that error propagates out of the returned future (callers that must not abort the batch should catch inside action and return a sentinel instead).

Implementation

Future<void> runWithConcurrency<T>(
  Iterable<T> items,
  int concurrency,
  Future<void> Function(T item) action,
) async {
  final sem = ChunkSemaphore(concurrency < 1 ? 1 : concurrency);
  final inflight = <Future<void>>[];
  for (final item in items) {
    await sem.acquire();
    inflight.add(() async {
      try {
        await action(item);
      } finally {
        sem.release();
      }
    }());
  }
  await Future.wait(inflight);
}