Pool future creation to ensure no more than 10 ops are in flight

pkg/barback/lib/src/utils.dart

Index: pkg/barback/lib/src/utils.dart
diff --git a/pkg/barback/lib/src/utils.dart b/pkg/barback/lib/src/utils.dart
index 95f498e037a634f3c0aac681663b242b57be6882..0bf4858bcc60ca2e54193d6e6714aef09cf17ece 100644
--- a/pkg/barback/lib/src/utils.dart
+++ b/pkg/barback/lib/src/utils.dart
@@ -196,3 +196,60 @@ Stream futureStream(Future<Stream> future) {
   });
   return controller.stream;
 }
+
+typedef Future<T> _FutureThing<T>(dynamic value);
+
+class _FuturePool<T> {
+  static const int _GROUP_SIZE = 10;
+
+  final Iterator<T> _iterator;
+  final _FutureThing<T> _factory;
+  final Completer _completer = new Completer();
+  int _runningCount = 0;
+
+  _FuturePool(this._iterator, this._factory) {
+    // TODO: runAsync?
+    _populate();
+  }
+
+  Future get future => _completer.future;
+
+  void _populate() {
+    bool itemsLeft = false;
+    while(_runningCount < _GROUP_SIZE && !_completer.isCompleted) {
+      itemsLeft = _iterator.moveNext();
+      if(!itemsLeft) break;
+
+      T item = _iterator.current;
+      Future future = _factory(item);
+      if(future != null) {
+        _runningCount++;
+        future.then(_onSuccess, onError: _onError);
+      }
+    }
+
+    assert(_runningCount >= 0);
+
+    if(!itemsLeft && _runningCount == 0 && !_completer.isCompleted) {
+      _completer.complete();
+    }
+  }
+
+  void _onSuccess(T value) {
+    assert(_runningCount > 0);
+    _runningCount--;
+    // TODO: runAsync?
+    _populate();
+  }
+
+  void _onError(Object err) {
+    assert(_runningCount > 0);
+    _runningCount--;
+    if(!_completer.isCompleted) _completer.completeError(err);
+  }
+}
+
+Future forEachPooledFuture(Iterable source, Future action(dynamic item)) {
+  var pool = new _FuturePool(source.iterator, action);
+  return pool.future;
+}