Expose aggregate transformers in barback.

pkg/barback/lib/src/graph/transform_node.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.transform_node;
 
 import 'dart:async';
 
 import '../asset/asset.dart';
 import '../asset/asset_id.dart';
 import '../asset/asset_node.dart';
 import '../asset/asset_node_set.dart';
 import '../errors.dart';
 import '../log.dart';
 import '../transformer/aggregate_transform.dart';
 import '../transformer/aggregate_transformer.dart';
 import '../transformer/declaring_aggregate_transform.dart';
 import '../transformer/declaring_aggregate_transformer.dart';
 import '../transformer/lazy_aggregate_transformer.dart';
 import '../utils.dart';
 import 'node_status.dart';
 import 'node_streams.dart';
 import 'phase.dart';
+import 'transformer_classifier.dart';
 
 /// Describes a transform on a set of assets and its relationship to the build
 /// dependency graph.
 ///
 /// Keeps track of whether it's dirty and needs to be run and which assets it
 /// depends on.
 class TransformNode {
   /// The aggregate key for this node.
   final String key;
 
+  /// The [TransformerClassifier] that [this] belongs to.
+  final TransformerClassifier classifier;
+
   /// The [Phase] that this transform runs in.
-  final Phase phase;
+  Phase get phase => classifier.phase;
 
   /// The [AggregateTransformer] to apply to this node's inputs.
   final AggregateTransformer transformer;
 
   /// The primary asset nodes this transform runs on.
   final _primaries = new AssetNodeSet();
 
   /// A string describing the location of [this] in the transformer graph.
   final String _location;
 
@@ skipping 109 matching lines @@
 
   /// The controller for the currently-running [AggregateTransformer.apply]
   /// call's [AggregateTransform].
   ///
   /// This will be non-`null` when [AggregateTransform.apply] is running, which
   /// means that it's always non-`null` when [_state] is [_State.APPLYING] or
   /// [_State.NEEDS_APPLY], sometimes non-`null` when it's
   /// [_State.NEEDS_DECLARE], and always `null` otherwise.
   AggregateTransformController _applyController;
 
-  TransformNode(this.phase, this.transformer, this.key, this._location) {
+  TransformNode(this.classifier, this.transformer, this.key, this._location) {
     _forced = transformer is! DeclaringAggregateTransformer;
 
     _phaseAssetSubscription = phase.previous.onAsset.listen((node) {
       if (!_missingInputs.contains(node.id)) return;
       if (_forced) node.force();
       _dirty();
     });
 
     _phaseStatusSubscription = phase.previous.onStatusChange.listen((status) {
       if (status == NodeStatus.RUNNING) return;
 
       _maybeFinishDeclareController();
       _maybeFinishApplyController();
     });
 
+    classifier.onDoneClassifying.listen((_) {
+      _maybeFinishDeclareController();
+      _maybeFinishApplyController();
+    });
+
     _run();
   }
 
   /// Adds [input] as a primary input for this node.
   void addPrimary(AssetNode input) {
     _primaries.add(input);
     if (_forced) input.force();
 
     _primarySubscriptions[input.id] = input.onStateChange
         .listen((_) => _onPrimaryStateChange(input));
 
     if (_state == _State.DECLARING && !_declareController.isDone) {
       // If we're running `declareOutputs` and its id stream isn't closed yet,
       // pass this in as another id.
       _declareController.addId(input.id);
       _maybeFinishDeclareController();
     } else if (_state == _State.APPLYING) {
       // If we're running `apply`, we need to wait until [input] is available
       // before we pass it into the stream. If it's available now, great; if
       // not, [_onPrimaryStateChange] will handle it.
-      if (!input.state.isAvailable) return;
+      if (!input.state.isAvailable) {
+        // If we started running eagerly without being forced, abort that run if
+        // a new unavailable asset comes in.
+        if (input.isLazy && !_forced) _restartRun();
+        return;
+      }
+
       _onPrimaryStateChange(input);
       _maybeFinishApplyController();
     } else {
       // Otherwise, a new input means we'll need to re-run `declareOutputs`.
       _restartRun();
     }
   }
 
   /// Marks this transform as removed.
   ///
@@ skipping 107 matching lines @@
         // re-run declare and then apply.
         _restartRun();
       } else {
         // If the new asset comes before the asset stream is done, we can just
         // pass it to the stream.
         _applyController.addInput(input.asset);
         _maybeFinishApplyController();
       }
     } else {
       if (_forced) input.force();
-      if (_state == _State.APPLYING && !_applyController.addedId(input.id)) {
+      if (_state == _State.APPLYING && !_applyController.addedId(input.id) &&
+          (_forced || !input.isLazy)) {
         // If the input hasn't yet been added to the transform's input stream,
-        // there's no need to consider the transformation dirty.
+        // there's no need to consider the transformation dirty. However, if the
+        // input is lazy and we're running eagerly, we need to restart the
+        // transformation.
         return;
       }
       _dirty();
     }
   }
 
   /// Run the entire transformation, including both `declareOutputs` (if
   /// applicable) and `apply`.
   void _run() {
     assert(_state != _State.DECLARING);
@@ skipping 40 matching lines @@
       return;
     }
 
     _state = _State.DECLARING;
     var controller = new DeclaringAggregateTransformController(this);
     _declareController = controller;
     _streams.onLogPool.add(controller.onLog);
     for (var primary in _primaries) {
       controller.addId(primary.id);
     }
+    _maybeFinishDeclareController();
 
     syncFuture(() {
       return (transformer as DeclaringAggregateTransformer)
           .declareOutputs(controller.transform);
     }).whenComplete(() {
       // Cancel the controller here even if `declareOutputs` wasn't interrupted.
       // Since the declaration is finished, we want to close out the
       // controller's streams.
       controller.cancel();
       _declareController = null;
@@ skipping 148 matching lines @@
   ///
   /// Returns whether or not an error occurred while running the transformer.
   Future<bool> _runApply() {
     var controller = new AggregateTransformController(this);
     _applyController = controller;
     _streams.onLogPool.add(controller.onLog);
     for (var primary in _primaries) {
       if (!primary.state.isAvailable) continue;
       controller.addInput(primary.asset);
     }
+    _maybeFinishApplyController();
 
     return syncFuture(() {
       return transformer.apply(controller.transform);
     }).whenComplete(() {
       // Cancel the controller here even if `apply` wasn't interrupted. Since
       // the apply is finished, we want to close out the controller's streams.
       controller.cancel();
       _applyController = null;
     }).then((_) {
       assert(_state != _State.DECLARED);
@@ skipping 114 matching lines @@
   void _consumePrimary(AssetId id) {
     var controller = _passThroughControllers.remove(id);
     if (controller == null) return;
     controller.setRemoved();
   }
 
   /// If `declareOutputs` is running and all previous phases have declared their
   /// outputs, mark [_declareController] as done.
   void _maybeFinishDeclareController() {
     if (_declareController == null) return;
+    if (classifier.isClassifying) return;
     if (phase.previous.status == NodeStatus.RUNNING) return;
     _declareController.done();
   }
 
   /// If `apply` is running, all previous phases have declared their outputs,
   /// and all primary inputs are available and thus have been passed to the
   /// transformer, mark [_applyController] as done.
   void _maybeFinishApplyController() {
     if (_applyController == null) return;
+    if (classifier.isClassifying) return;
     if (_primaries.any((input) => !input.state.isAvailable)) return;
     if (phase.previous.status == NodeStatus.RUNNING) return;
     _applyController.done();
   }
 
   BarbackException _wrapException(error, StackTrace stackTrace) {
     if (error is! AssetNotFoundException) {
       return new TransformerException(info, error, stackTrace);
     } else {
       return new MissingInputException(info, error.id);
@@ skipping 61 matching lines @@
   /// declaring and [APPLYING] otherwise. If a primary input is added or
   /// removed, this will transition to [DECLARING].
   static const APPLIED = const _State._("applied");
 
   final String name;
 
   const _State._(this.name);
 
   String toString() => name;
 }