throttled_iterator.dart

throttled_iterator.dart

extension on Iterator {
  void throttledForEach(int parallelFutures, ForEachCallback onData,
      [ItemProcessedCallback onProgress]) async {
    final semaphores = StreamController<int>();
    while (parallelFutures-- > 0) {
      semaphores.add(1);
    }
    await for (var _ in semaphores.stream) {
      if (!moveNext()) break;
      var f = onData(current)
        ..then((value) => semaphores.add(1))
        ..catchError((e) {
          semaphores.add(1);
          throw e;
        });
      if (onProgress != null) {
        f.then((value) => onProgress);
      }
      await f;
    }
  }
}