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

pkg/barback/lib/src/phase_input.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.phase_input;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'asset.dart';
@@ skipping 165 matching lines @@
     });
 
     // Don't top-level errors coming from the input processing. Any errors will
     // eventually be piped through [process]'s returned Future.
     _adjustTransformersFuture.catchError((_) {});
   }
 
   // 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;
 
         // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
         // results.
         return transform.transformer.isPrimary(asset);
       }).then((isPrimary) {
         if (isPrimary) return;
         _transforms.remove(transform);
         transform.remove();
       });
-    }));
+    });
   }
 
   // Add new transforms for transformers that consider [input]'s asset to be a
   // primary input.
   //
   // [oldTransformers] is the set of transformers for which there were
   // transforms that had [input] as a primary input prior to this. They don't
   // need to be checked, since their transforms were removed or preserved in
   // [_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
       // re-run.
       // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
       // results.
       return transformer.isPrimary(input.asset).then((isPrimary) {
         if (!isPrimary) return;
         var transform = new TransformNode(_phase, transformer, input);
         _transforms.add(transform);
         _onDirtyPool.add(transform.onDirty);
       });
-    }));
+    });
   }
 
   /// Adjust whether [input] is passed through the phase unmodified, based on
   /// whether it's consumed by other transforms in this phase.
   ///
   /// If [input] was already passed-through, this will update the passed-through
   /// value.
   void _adjustPassThrough() {
     assert(input.state.isAvailable);
 
@@ skipping 42 matching lines @@
   Future<Set<AssetNode>> _processTransforms() {
     if (input.state.isRemoved) return new Future.value(new Set());
 
     if (_passThroughController != null) {
       if (!_newPassThrough) return new Future.value(new Set());
       _newPassThrough = false;
       return new Future.value(
           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));
   }
 }