Add aggregate transformer types to barback.

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

+// Copyright (c) 2014, 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.transformer.declaring_aggregate_transform;
+
+import 'dart:async';
+
+import '../asset/asset_id.dart';
+import '../graph/transform_node.dart';
+import 'base_transform.dart';
+
+/// A transform for [DeclaringAggregateTransformer]s that allows them to declare
+/// the ids of the outputs they'll generate without generating the concrete
+/// bodies of those outputs.
+class DeclaringAggregateTransform extends BaseTransform {

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

A bunch of my comments on AggregateTransform apply here too.

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

Done.

+  /// The set of output ids declared by the transformer.
+  final _outputIds = new Set<AssetId>();
+
+  /// The aggregate key for this transform.
+  final key;
+
+  /// The stream of primary input ids that have been aggregated for this
+  /// transform.
+  ///
+  /// This is exposed as a stream so that the transformer can start working
+  /// before all its input ids 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.
+  ///
+  /// A transformer may complete its `declareOutputs` 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.primaryIds.take(2)` to
+  /// access only those inputs' ids.

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

This seems like it would be more helpful as a doc comment for
DeclaringTransformer.declareOutputs() instead of here.

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

I'll put it both places.

+  Stream<AssetId> get primaryIds => _primaryIds;
+  Stream<AssetId> _primaryIds;
+
+  /// The controller for [primaryIds].
+  ///
+  /// This is a broadcast controller so that the transform can keep
+  /// [_allPrimaryIds] up to date.
+  final _idController = new StreamController<AssetId>();
+
+  /// The set of all primary input ids that have been emitted by [primaryIds].
+  final _allPrimaryIds = new Set<AssetId>();
+
+  DeclaringAggregateTransform._(TransformNode node)
+      : key = node.key,
+        super(node) {
+    _idController.stream.listen(_allPrimaryId.add);

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

_allPrimaryIds

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

Done.

+    // [primaryIds] should be a non-broadcast stream.
+    _primaryIds = broadcastToSingleSubscription(_idController.stream);
+  }
+
+  /// Stores [id] as the id of an output that will be created by this
+  /// transformation when it's run.
+  ///
+  /// A transformation can declare as many assets as it wants. If
+  /// [DeclaringTransformer.declareOutputs] declareds a given asset id for a

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

"declareds" -> "declares".

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

Done.

+  /// given input, [Transformer.apply] should emit the corresponding asset as
+  /// well.
+  void declareOutput(AssetId id) {
+    // TODO(nweiz): This should immediately throw if an output with that ID
+    // has already been declared by this transformer.

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

Is there a reason we want to throw for this instead of just ignoring it?

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

It's effectively a collision—the user has provided two separate outputs with the
same id, which is an error anywhere else. The benefit is that here we can
complain about it immediately when it happens.

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

I suppose so, but part of the reason we care about asset collisions is because
you could output two *different* assets with the same ID. There's no right way
to handle that.

In this case, it's just a pure ID, so outputting it twice doesn't cause any loss
of information.

I don't have a strong preference here, though, so do it how you like.

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

In the non-declaring transformer, though, it's not a pure ID, and I think it's
nice to have behavioral symmetry here.

+    _outputIds.add(id);
+  }
+
+  void consumePrimary(AssetId id) {
+    if (!_allPrimaryIds.contains(id)) {
+      throw new StateError(
+          "$id can't be consumed because it's not a primary input.");
+    }
+
+    super();
+  }
+}
+
+/// The controller for [DeclaringAggregateTransform].
+class DeclaringAggregateTransformController extends BaseTransformController {
+  DeclaringAggregateTransform get transform => super.transform;
+
+  /// The set of ids that the transformer declares it will emit.
+  Set<AssetId> get outputIds => transform._outputIds;
+
+  /// The controller for the [DeclaringAggregateTransform.primaryIds] stream.
+  StreamController<AssetId> get idController => transform._idController;
+
+  DeclaringAggregateTransformController(TransformNode node)
+      : super(new DeclaringAggregateTransform._(node));
+
+  void close() {
+    super.close();
+    idController.close();
+  }
+}