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

pkg/barback/lib/src/phase_input.dart

Index: pkg/barback/lib/src/phase_input.dart
diff --git a/pkg/barback/lib/src/phase_input.dart b/pkg/barback/lib/src/phase_input.dart
index d5945d921b6e9e413336e8d2bfa020a5840059b0..88fee0b89408fc25e435136f07c07216d365d063 100644
--- a/pkg/barback/lib/src/phase_input.dart
+++ b/pkg/barback/lib/src/phase_input.dart
@@ -183,7 +183,10 @@ class PhaseInput {
   // Remove any old transforms that used to have [asset] as a primary asset but
   // no longer apply to its new contents.
   Future _removeStaleTransforms(Asset asset, Set<Transformer> transformers) {
-    return Future.wait(_transforms.map((transform) {
+
+    var txCopy = _transforms.toList();
+
+    return forEachPooledFuture(txCopy, (transform) {
       return newFuture(() {
         if (!transformers.contains(transform.transformer)) return false;
 
@@ -195,7 +198,7 @@ class PhaseInput {
         _transforms.remove(transform);
         transform.remove();
       });
-    }));
+    });
   }
 
   // Add new transforms for transformers that consider [input]'s asset to be a
@@ -207,12 +210,13 @@ class PhaseInput {
   // [_removeStaleTransforms].
   Future _addFreshTransforms(Set<Transformer> transformers,
       Set<Transformer> oldTransformers) {
-    return Future.wait(transformers.map((transformer) {
-      if (oldTransformers.contains(transformer)) return new Future.value();
+
+    return forEachPooledFuture(transformers.toList(), (transformer) {
+      if (oldTransformers.contains(transformer)) return null;
 
       // If the asset is unavailable, the results of this [_adjustTransformers]
       // run will be discarded, so we can just short-circuit.
-      if (input.asset == null) return new Future.value();
+      if (input.asset == null) return null;
 
       // We can safely access [input.asset] here even though it might have
       // changed since (as above) if it has, [_adjustTransformers] will just be
@@ -225,7 +229,7 @@ class PhaseInput {
         _transforms.add(transform);
         _onDirtyPool.add(transform.onDirty);
       });
-    }));
+    });
   }
 
   /// Adjust whether [input] is passed through the phase unmodified, based on
@@ -288,9 +292,12 @@ class PhaseInput {
           new Set<AssetNode>.from([_passThroughController.node]));
     }
 
-    return Future.wait(_transforms.map((transform) {
+    var outputs = new List<Set>();
+
+    return forEachPooledFuture(_transforms.toList(), (transform) {
       if (!transform.isDirty) return new Future.value(new Set());
-      return transform.apply();
-    })).then((outputs) => unionAll(outputs));
+      return transform.apply()
+        .then(outputs.add);
+    }).then((_) => unionAll(outputs));
   }
 }