Add aggregate transformer types to 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 '../errors.dart';
 import '../log.dart';
+import '../transformer/aggregate_transform.dart';
+import '../transformer/declaring_aggregate_transform.dart';
 import '../transformer/declaring_transform.dart';
 import '../transformer/declaring_transformer.dart';
 import '../transformer/lazy_transformer.dart';
 import '../transformer/transform.dart';
 import '../transformer/transformer.dart';
-import '../utils.dart';
 import 'node_status.dart';
 import 'node_streams.dart';
 import 'phase.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 [Phase] that this transform runs in.
   final Phase phase;
 
   /// The [Transformer] to apply to this node's inputs.
   final Transformer transformer;
 
   /// The node for the primary asset this transform depends on.
   final AssetNode primary;
 
   /// A string describing the location of [this] in the transformer graph.
@@ skipping 79 matching lines @@
   bool _consumePrimary = false;
 
   /// The set of output ids that [transformer] declared it would emit.
   ///
   /// This is only non-null if [transformer] is a [DeclaringTransformer] and its
   /// [declareOutputs] has been run successfully.
   Set<AssetId> _declaredOutputs;
 
   TransformNode(this.phase, Transformer transformer, AssetNode primary,
       this._location)
-      : transformer = transformer,
+      : key = primary.id.path,
+        transformer = transformer,
         primary = primary {
     _forced = transformer is! DeclaringTransformer;
     if (_forced) primary.force();
 
     _primarySubscription = primary.onStateChange.listen((state) {
       if (state.isRemoved) {
         remove();
       } else {
         if (_forced) primary.force();
         _dirty();
@@ skipping 86 matching lines @@
     } else {
       _state = _State.NEEDS_APPLY;
     }
   }
 
   /// Runs [transform.declareOutputs] and emits the resulting assets as dirty
   /// assets.
   Future _declareOutputs() {
     if (transformer is! DeclaringTransformer) return new Future.value();
 
-    var controller = new DeclaringTransformController(this);
-    return syncFuture(() {
-      return (transformer as DeclaringTransformer)
-          .declareOutputs(controller.transform);
+    var controller = new DeclaringAggregateTransformController(this);
+    _streams.onLogPool.add(controller.onLog);
+    controller.idController.add(primary.id);
+    return newDeclaringTransform(controller.transform).then((transform) {
+      return (transformer as DeclaringTransformer).declareOutputs(transform);
     }).then((_) {
       if (_isRemoved) return;
       if (controller.loggedError) return;
 
-      _consumePrimary = controller.consumePrimary;
+      _consumePrimary = controller.consumedPrimaries.contains(primary.id);
       _declaredOutputs = controller.outputIds;
       var invalidIds = _declaredOutputs
           .where((id) => id.package != phase.cascade.package).toSet();
       for (var id in invalidIds) {
         _declaredOutputs.remove(id);
         // TODO(nweiz): report this as a warning rather than a failing error.
         phase.cascade.reportError(new InvalidOutputException(info, id));
       }
 
       if (!_declaredOutputs.contains(primary.id)) _emitPassThrough();
@@ skipping 82 matching lines @@
       });
 
       return node.asset;
     });
   }
 
   /// Run [Transformer.apply] as soon as [primary] is available.
   ///
   /// Returns whether or not an error occurred while running the transformer.
   Future<bool> _runApply() {
-    var transformController = new TransformController(this);
-    _streams.onLogPool.add(transformController.onLog);
+    var controller = new AggregateTransformController(this);
+    _streams.onLogPool.add(controller.onLog);
 
     return primary.whenAvailable((_) {
       if (_isRemoved) return null;
       _state = _State.APPLYING;
-      return syncFuture(() => transformer.apply(transformController.transform));
+      controller.inputController.add(primary.asset);
+      return newTransform(controller.transform).then((transform) {
+        return transformer.apply(transform);
+      });
     }).then((_) {
       if (!_forced && !primary.state.isAvailable) {
         _state = _State.DECLARED;
         _streams.changeStatus(NodeStatus.IDLE);
         return false;
       }
 
       if (_isRemoved) return false;
       if (_state == _State.NEEDS_APPLY) return false;
       if (_state == _State.DECLARING) return false;
-      if (transformController.loggedError) return true;
-      _handleApplyResults(transformController);
+      if (controller.loggedError) return true;
+      _handleApplyResults(controller);
       return false;
     }).catchError((error, stackTrace) {
       // If the transform became dirty while processing, ignore any errors from
       // it.
       if (_state == _State.NEEDS_APPLY || _isRemoved) return false;
 
       // Catch all transformer errors and pipe them to the results stream. This
       // is so a broken transformer doesn't take down the whole graph.
       phase.cascade.reportError(_wrapException(error, stackTrace));
       return true;
     });
   }
 
   /// Handle the results of running [Transformer.apply].
   ///
-  /// [transformController] should be the controller for the [Transform] passed
-  /// to [Transformer.apply].
-  void _handleApplyResults(TransformController transformController) {
-    _consumePrimary = transformController.consumePrimary;
+  /// [controller] should be the controller for the [AggegateTransform] passed
+  /// to [AggregateTransformer.apply].
+  void _handleApplyResults(AggregateTransformController controller) {
+    _consumePrimary = controller.consumedPrimaries.contains(primary.id);
 
-    var newOutputs = transformController.outputs;
+    var newOutputs = controller.outputs;
     // Any ids that are for a different package are invalid.
     var invalidIds = newOutputs
         .map((asset) => asset.id)
         .where((id) => id.package != phase.cascade.package)
         .toSet();
     for (var id in invalidIds) {
       newOutputs.removeId(id);
       // TODO(nweiz): report this as a warning rather than a failing error.
       phase.cascade.reportError(new InvalidOutputException(info, id));
     }
@@ skipping 129 matching lines @@
   /// If an input changes, this will transition to [DECLARED] if the transform
   /// is deferred and [APPLYING] otherwise.
   static final APPLIED = const _State._("applied");
 
   final String name;
 
   const _State._(this.name);
 
   String toString() => name;
 }