Roll to pickup pool changes

packages/barback/lib/src/graph/phase.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.graph.phase;
 
 import 'dart:async';
 
 import 'package:collection/collection.dart';
 
@@ skipping 74 matching lines @@
   final _streams = new NodeStreams();
   Stream<NodeStatus> get onStatusChange => _streams.onStatusChange;
   Stream<AssetNode> get onAsset => _streams.onAsset;
   Stream<LogEntry> get onLog => _streams.onLog;
 
   /// How far along [this] is in processing its assets.
   NodeStatus get status {
     // Before any transformers are added, the phase should be dirty if and only
     // if any input is dirty.
     if (_classifiers.isEmpty && _groups.isEmpty && previous == null) {
-      return _inputs.any((input) => input.state.isDirty) ?
-          NodeStatus.RUNNING : NodeStatus.IDLE;
+      return _inputs.any((input) => input.state.isDirty)
+          ? NodeStatus.RUNNING
+          : NodeStatus.IDLE;
     }
 
-    var classifierStatus = NodeStatus.dirtiest(
-        _classifiers.values.map((classifier) => classifier.status));
-    var groupStatus = NodeStatus.dirtiest(
-        _groups.values.map((group) => group.status));
+    var classifierStatus = NodeStatus
+        .dirtiest(_classifiers.values.map((classifier) => classifier.status));
+    var groupStatus =
+        NodeStatus.dirtiest(_groups.values.map((group) => group.status));
     return (previous == null ? NodeStatus.IDLE : previous.status)
         .dirtier(classifierStatus)
         .dirtier(groupStatus);
   }
 
   /// The previous phase in the cascade, or null if this is the first phase.
   final Phase previous;
 
   /// The subscription to [previous]'s [onStatusChange] stream.
   StreamSubscription _previousStatusSubscription;
@@ skipping 14 matching lines @@
   Set<AssetNode> get availableOutputs {
     return _outputs.values
         .map((output) => output.output)
         .where((node) => node.state.isAvailable)
         .toSet();
   }
 
   // TODO(nweiz): Rather than passing the cascade and the phase everywhere,
   // create an interface that just exposes [getInput]. Emit errors via
   // [AssetNode]s.
-  Phase(AssetCascade cascade, String location)
-      : this._(cascade, location, 0);
+  Phase(AssetCascade cascade, String location) : this._(cascade, location, 0);
 
   Phase._(this.cascade, this._location, this._index, [this.previous]) {
     if (previous != null) {
       _previousOnAssetSubscription = previous.onAsset.listen(addInput);
-      _previousStatusSubscription = previous.onStatusChange
-          .listen((_) => _streams.changeStatus(status));
+      _previousStatusSubscription =
+          previous.onStatusChange.listen((_) => _streams.changeStatus(status));
     }
 
     onStatusChange.listen((status) {
       if (status == NodeStatus.RUNNING) return;
 
       // All the previous phases have finished declaring or producing their
       // outputs. If anyone's still waiting for outputs, cut off the wait; we
       // won't be generating them, at least until a source asset changes.
       for (var completer in _pendingOutputRequests.values) {
         completer.complete(null);
       }
       _pendingOutputRequests.clear();
     });
   }
 
   /// Adds a new asset as an input for this phase.
   ///
   /// [node] doesn't have to be [AssetState.AVAILABLE]. Once it is, the phase
   /// will automatically begin determining which transforms can consume it as a
   /// primary input. The transforms themselves won't be applied until [process]
   /// is called, however.
   ///
   /// This should only be used for brand-new assets or assets that have been
   /// removed and re-created. The phase will automatically handle updated assets
   /// using the [AssetNode.onStateChange] stream.
   void addInput(AssetNode node) {
     // Each group is one channel along which an asset may be forwarded, as is
     // each transformer.
-    var forwarder = new PhaseForwarder(
-        node, _classifiers.length, _groups.length);
+    var forwarder =
+        new PhaseForwarder(node, _classifiers.length, _groups.length);
     _forwarders[node.id] = forwarder;
     forwarder.onAsset.listen(_handleOutputWithoutForwarder);
     if (forwarder.output != null) {
       _handleOutputWithoutForwarder(forwarder.output);
     }
 
     _inputOrigins.add(node.origin);
     _inputs.add(node);
     _inputSubscriptions.add(node.onStateChange.listen((state) {
       if (state.isRemoved) {
@@ skipping 11 matching lines @@
   // TODO(nweiz): If the output is available when this is called, it's
   // theoretically possible for it to become unavailable between the call and
   // the return. If it does so, it won't trigger the rebuilding process. To
   // avoid this, we should have this and the methods it calls take explicit
   // callbacks, as in [AssetNode.whenAvailable].
   /// Gets the asset node for an output [id].
   ///
   /// If [id] is for a generated or transformed asset, this will wait until it
   /// has been created and return it. This means that the returned asset will
   /// always be [AssetState.AVAILABLE].
-  /// 
+  ///
   /// If the output cannot be found, returns null.
   Future<AssetNode> getOutput(AssetId id) {
     return new Future.sync(() {
       if (id.package != cascade.package) return cascade.graph.getAssetNode(id);
       if (_outputs.containsKey(id)) {
         var output = _outputs[id].output;
         // If the requested output is available, we can just return it.
         if (output.state.isAvailable) return output;
 
         // If the requested output exists but isn't yet available, wait to see
         // if it becomes available. If it's removed before becoming available,
         // try again, since it could be generated again.
         output.force();
         return output.whenAvailable((_) {
           return output;
         }).catchError((error) {
           if (error is! AssetNotFoundException) throw error;
           return getOutput(id);
         });
       }
 
       // If this phase and the previous phases are fully declared or done, the
       // requested output won't be generated and we can safely return null.
       if (status != NodeStatus.RUNNING) return null;
 
       // Otherwise, store a completer for the asset node. If it's generated in
       // the future, we'll complete this completer.
-      var completer = _pendingOutputRequests.putIfAbsent(id,
-          () => new Completer.sync());
+      var completer =
+          _pendingOutputRequests.putIfAbsent(id, () => new Completer.sync());
       return completer.future;
     });
   }
 
   /// Set this phase's transformers to [transformers].
   void updateTransformers(Iterable transformers) {
     var newTransformers = transformers
         .where((op) => op is Transformer || op is AggregateTransformer)
         .toSet();
     var oldTransformers = _classifiers.keys.toSet();
     for (var removed in oldTransformers.difference(newTransformers)) {
       _classifiers.remove(removed).remove();
     }
 
     for (var transformer in newTransformers.difference(oldTransformers)) {
-      var classifier = new TransformerClassifier(
-          this, transformer, "$_location.$_index");
+      var classifier =
+          new TransformerClassifier(this, transformer, "$_location.$_index");
       _classifiers[transformer] = classifier;
       classifier.onAsset.listen(_handleOutput);
       _streams.onLogPool.add(classifier.onLog);
       classifier.onStatusChange.listen((_) => _streams.changeStatus(status));
       for (var input in _inputs) {
         classifier.addInput(input);
       }
     }
 
-    var newGroups = DelegatingSet.typed/*<TransformerGroup>*/(
+    var newGroups = DelegatingSet.typed<TransformerGroup>(
         transformers.where((op) => op is TransformerGroup).toSet());
     var oldGroups = _groups.keys.toSet();
     for (var removed in oldGroups.difference(newGroups)) {
       _groups.remove(removed).remove();
     }
 
     for (var added in newGroups.difference(oldGroups)) {
       var runner = new GroupRunner(previous, added, "$_location.$_index");
       _groups[added] = runner;
       runner.onAsset.listen(_handleOutput);
@@ skipping 74 matching lines @@
     }
   }
 
   /// Add [asset] as an output of this phase without checking if it's a
   /// forwarded asset.
   void _handleOutputWithoutForwarder(AssetNode asset) {
     if (_outputs.containsKey(asset.id)) {
       _outputs[asset.id].add(asset);
     } else {
       _outputs[asset.id] = new PhaseOutput(this, asset, "$_location.$_index");
-      _outputs[asset.id].onAsset.listen(_emit,
-          onDone: () => _outputs.remove(asset.id));
+      _outputs[asset.id]
+          .onAsset
+          .listen(_emit, onDone: () => _outputs.remove(asset.id));
       _emit(_outputs[asset.id].output);
     }
 
     var exception = _outputs[asset.id].collisionException;
     if (exception != null) cascade.reportError(exception);
   }
 
   /// Emit [asset] as an output of this phase.
   ///
   /// This should be called after [_handleOutput], so that collisions are
@@ skipping 11 matching lines @@
 
     if (asset.state.isAvailable) {
       request.complete(asset);
       return;
     }
 
     // A lazy asset may be emitted while still dirty. If so, we wait until it's
     // either available or removed before trying again to access it.
     assert(asset.state.isDirty);
     asset.force();
-    asset.whenStateChanges().then((state) {
-      if (state.isRemoved) return getOutput(asset.id);
-      return asset;
-    })
-       .then((asset) => request.complete(asset))
-       .catchError(request.completeError);
+    asset
+        .whenStateChanges()
+        .then((state) {
+          if (state.isRemoved) return getOutput(asset.id);
+          return asset;
+        })
+        .then((asset) => request.complete(asset))
+        .catchError(request.completeError);
   }
 
   String toString() => "phase $_location.$_index";
 }