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 51%
copy from pkg/barback/lib/src/transformer/transform.dart
copy to pkg/barback/lib/src/transformer/aggregate_transform.dart
index 809cd122683484ae84c717edd0da3e1b77b9d851..04d978f78bb973e70b65a04ede97ae1766d19bd3 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,57 @@ 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 aggregate key for this transform.

[Bob Nystrom, 2014/05/05 23:41:56]

"Aggregate key" already means something to a lot of programmers (A key in SQL
made of more than one column), so this is confusing to me. What do you think of
just calling it a "transform key"? The Transform class already represents a
single application of a transformer to a set of inputs, so I think it's the
right idea.

[nweiz, 2014/05/06 22:46:40]

Done.

+  get key => _node.key;
+
+  /// The stream of primary inputs that have been aggregated for this transform.

[Bob Nystrom, 2014/05/05 23:41:56]

"have been aggregated for" -> "will be processed by"?

[nweiz, 2014/05/06 22:46:40]

Done.

   ///
-  /// 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 exposed as a stream so that the transformer can start working
+  /// before all its inputs are available. The stream will be closed once
+  /// barback determines that no more inputs exist or will be generated for this
+  /// transform without external modification.

[Bob Nystrom, 2014/05/05 23:41:56]

This is a bit hard to read in part because of the future tense. How about:

"The stream is closed when all inputs have been provided."

[nweiz, 2014/05/06 22:46:40]

I want to be explicit that the stream will stay open even after all assets have
been provided since it might still take some time to prove that there are no
more assets.

[Bob Nystrom, 2014/05/06 23:55:09]

Maybe just say that then: "The stream is closed not just when all inputs are
provided, but when barback is sure no more inputs will are forthcoming."

[nweiz, 2014/05/07 01:28:50]

Done.

   ///
-  /// 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.
+  /// 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.

[Bob Nystrom, 2014/05/05 23:41:56]

I think this is a bit misleading. If the transformer knows it will only have two
inputs, it should really do transform.primaryInputs.toList() and assert the
length is two. Doing take(2) communicates "there may be more but I only want the
first two", which isn't safe in barback since you have no idea what order your
primaries will appear.

[nweiz, 2014/05/06 22:46:40]

`toList()` is definitely not what you want to do in this situation, since it
will block unnecessarily until barback can prove that there are no additional
assets to be processed. Unfortunately, there's no method on Stream that says
"take N and assert that nothing else is emitted", but I think `take(2)` is a
fine way to write this.

[Bob Nystrom, 2014/05/06 23:55:09]

Oof, we should discuss this, because I think that definitely what users will
naturally think to do. If that's the wrong practice, we're setting them up for
failure.

I do understand what you're saying, I just wonder if there's some API tweaks we
can do to make this more intuitive.

[nweiz, 2014/05/07 01:28:50]

toList() will work just fine for 99% of aggregate transformers. I doubt many
aggregate transformers (other than auto-wrapped Transformers) will actually want
to consume a finite, known number of assets. I don't think we need to go out of
our way to make sure that they'll always do so in the most optimal way.

+  Stream<Asset> get primaryInputs => _primaryInputs;
+  Stream<Asset> _primaryInputs;
+
+  /// The controller for [primaryInputs].
   ///
-  /// 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;
+  /// This is a broadcast controller so that the transform can keep
+  /// [_allPrimaryInputs] up to date.
+  final _inputController = new StreamController<Asset>.broadcast();
+
+  /// The set of all primary inputs that have been emitted by [primaryInputs].
+  final _allPrimaryInputs = new AssetSet();

[Bob Nystrom, 2014/05/05 23:41:56]

"all" isn't very helpful here. How about "_emittedPrimaryInputs"?

[nweiz, 2014/05/06 22:46:40]

Done.

+
+  AggregateTransform._(TransformNode node)
+      : _node = node,
+        super(node) {
+    _inputController.stream.listen(_allPrimaryInputs.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 (_allPrimaryInputs.containsId(id)) {
+      return syncFuture(() => _allPrimaryInputs[id]);
+    } else {
+      return _node.getInput(id);
+    }
   }
 
   /// A convenience method to the contents of the input with [id] as a string.
@@ -105,15 +111,32 @@ class Transform extends BaseTransform {
     // has already been created by this transformer.
     _outputs.add(output);
   }
+
+  void consumePrimary(AssetId id) {
+    if (!_allPrimaryInputs.containsId(id)) {

[Bob Nystrom, 2014/05/05 23:41:56]

What if you know what the primary input *will* be and want to consume it even
though it hasn't come through the stream yet?

[nweiz, 2014/05/06 22:46:40]

You really shouldn't be assuming that an asset exists without having been passed
it explicitly; that's a recipe for failure.

Even if there are some narrow cases where this is possible to do safely, I think
providing useful and immediate feedback when someone's doing something wrong is
more important than supporting them.

[Bob Nystrom, 2014/05/06 23:55:09]

SGTM.

+      throw new StateError(
+          "$id can't be consumed because it's not a primary input.");
+    }
+
+    super();
+  }
 }
 
-/// 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();
+    inputController.close();
+  }
 }