Add aggregate transformer types to barback.

pkg/barback/lib/src/transformer/aggregate_transform.dart

Index: pkg/barback/lib/src/transformer/aggregate_transform.dart
diff --git a/pkg/barback/lib/src/transformer/transform.dart b/pkg/barback/lib/src/transformer/aggregate_transform.dart
similarity index 50%
copy from pkg/barback/lib/src/transformer/transform.dart
copy to pkg/barback/lib/src/transformer/aggregate_transform.dart
index 809cd122683484ae84c717edd0da3e1b77b9d851..a5244b43b0626a170e0fba86d8a2d75a565aac7a 100644
--- a/pkg/barback/lib/src/transformer/transform.dart
+++ b/pkg/barback/lib/src/transformer/aggregate_transform.dart
@@ -2,7 +2,7 @@
 // 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.transformer.transform;
+library barback.transformer.aggregate_transform;
 
 import 'dart:async';
 import 'dart:convert';
@@ -15,51 +15,60 @@ import '../graph/transform_node.dart';
 import '../utils.dart';
 import 'base_transform.dart';
 
-/// While a [Transformer] represents a *kind* of transformation, this defines
-/// one specific usage of it on a set of files.
-///
-/// This ephemeral object exists only during an actual transform application to
-/// facilitate communication between the [Transformer] and the code hosting
-/// the transformation. It lets the [Transformer] access inputs and generate
-/// outputs.
-class Transform extends BaseTransform {
+/// A transform for [AggregateTransformer]s that provides access to all of their
+/// primary inputs.
+class AggregateTransform extends BaseTransform {
   final TransformNode _node;
 
+  /// The set of outputs emitted by the transformer.
   final _outputs = new AssetSet();
 
-  /// Gets the primary input asset.
+  /// The transform key.
   ///
-  /// While a transformation can use multiple input assets, one must be a
-  /// special "primary" asset. This will be the "entrypoint" or "main" input
-  /// file for a transformation.
+  /// This is the key returned by [AggregateTransformer.classifyPrimary] for all
+  /// the assets in this transform.
+  String get key => _node.key;
+
+  /// The stream of primary inputs that will be processed by this transform.
   ///
-  /// For example, with a dart2js transform, the primary input would be the
-  /// entrypoint Dart file. All of the other Dart files that that imports
-  /// would be secondary inputs.
+  /// This is exposed as a stream so that the transformer can start working
+  /// before all its inputs are available. The stream is closed not just when
+  /// all inputs are provided, but when barback is confident no more inputs will
+  /// be forthcoming.
   ///
-  /// This method may fail at runtime with an [AssetNotFoundException] if called
-  /// asynchronously after the transform begins running. The primary input may
-  /// become unavailable while this transformer is running due to asset changes
-  /// earlier in the graph. You can ignore the error if this happens: the
-  /// transformer will be re-run automatically for you.
-  Asset get primaryInput {
-    if (!_node.primary.state.isAvailable) {
-      throw new AssetNotFoundException(_node.primary.id);
-    }
-
-    return _node.primary.asset;
+  /// A transformer may complete its `apply` method before this stream is
+  /// closed. For example, it may know that each key will only have two inputs
+  /// associated with it, and so use `transform.primaryInputs.take(2)` to access
+  /// only those inputs.
+  Stream<Asset> get primaryInputs => _primaryInputs;
+  Stream<Asset> _primaryInputs;
+
+  /// The controller for [primaryInputs].
+  ///
+  /// This is a broadcast controller so that the transform can keep
+  /// [_emittedPrimaryInputs] up to date.
+  final _inputController = new StreamController<Asset>.broadcast();
+
+  /// The set of all primary inputs that have been emitted by [primaryInputs].
+  final _emittedPrimaryInputs = new AssetSet();
+
+  AggregateTransform._(TransformNode node)
+      : _node = node,
+        super(node) {
+    _inputController.stream.listen(_emittedPrimaryInputs.add);
+    // [primaryInputs] should be a non-broadcast stream.
+    _primaryInputs = broadcastToSingleSubscription(_inputController.stream);
   }
 
-  Transform._(TransformNode node)
-    : _node = node,
-      super(node);
-
   /// Gets the asset for an input [id].
   ///
   /// If an input with [id] cannot be found, throws an [AssetNotFoundException].
   Future<Asset> getInput(AssetId id) {
-    if (id == _node.primary.id) return syncFuture(() => primaryInput);
-    return _node.getInput(id);
+    if (_emittedPrimaryInputs.containsId(id)) {
+      return syncFuture(() => _emittedPrimaryInputs[id]);
+    } else {
+      return _node.getInput(id);
+    }
   }
 
   /// A convenience method to the contents of the input with [id] as a string.
@@ -105,15 +114,32 @@ class Transform extends BaseTransform {
     // has already been created by this transformer.
     _outputs.add(output);
   }
+
+  void consumePrimary(AssetId id) {
+    if (!_emittedPrimaryInputs.containsId(id)) {
+      throw new StateError(
+          "$id can't be consumed because it's not a primary input.");
+    }
+
+    super.consumePrimary(id);
+  }
 }
 
-/// The controller for [Transform].
-class TransformController extends BaseTransformController {
-  Transform get transform => super.transform;
+/// The controller for [AggregateTransform].
+class AggregateTransformController extends BaseTransformController {
+  AggregateTransform get transform => super.transform;
 
   /// The set of assets that the transformer has emitted.
   AssetSet get outputs => transform._outputs;
 
-  TransformController(TransformNode node)
-      : super(new Transform._(node));
+  /// The controller for the [AggregateTransform.primaryInputs] stream.
+  StreamController<Asset> get inputController => transform._inputController;
+
+  AggregateTransformController(TransformNode node)
+      : super(new AggregateTransform._(node));
+
+  void close() {
+    super.close();
+    transform._inputController.close();
+  }
 }