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

pkg/barback/lib/src/utils.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 library barback.utils;
 
 import 'dart:async';
 
 /// A pair of values.
 class Pair<E, F> {
@@ skipping 178 matching lines @@
     stream.listen(
         controller.add,
         onError: (error) => controller.addError(error),
         onDone: controller.close);
   }).catchError((e) {
     controller.addError(e);
     controller.close();
   });
   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;
+}