Add support for transformer clusters to barback.

pkg/barback/lib/src/cluster_runner.dart

Index: pkg/barback/lib/src/cluster_runner.dart
diff --git a/pkg/barback/lib/src/cluster_runner.dart b/pkg/barback/lib/src/cluster_runner.dart
new file mode 100644
index 0000000000000000000000000000000000000000..6a58ba9b69f527c14e49e8006402770d59961be3
--- /dev/null
+++ b/pkg/barback/lib/src/cluster_runner.dart
@@ -0,0 +1,84 @@
+// 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.cluster_runner;
+
+import 'dart:async';
+
+import 'asset_cascade.dart';
+import 'asset_node.dart';
+import 'phase.dart';
+import 'stream_pool.dart';
+import 'transformer_cluster.dart';
+
+/// A class that handles a single transformer cluster.

[Bob Nystrom, 2013/10/04 21:47:50]

"handles" -> "processes all of the phases in"

[nweiz, 2013/10/07 23:21:31]

Done.

+///
+/// A cluster takes a many inputs, processes them, and emits many outputs.

[Bob Nystrom, 2013/10/04 21:47:50]

"a many" -> "many".

[nweiz, 2013/10/07 23:21:31]

Done.

+class ClusterRunner {
+  /// The phases defined by this cluster.
+  final _phases = new List<Phase>();
+
+  /// A stream that emits an event whenever this cluster becomes dirty and needs
+  /// to be run.
+  ///
+  /// This may emit events when the cluster was already dirty or while
+  /// processing transforms. Events are emitted synchronously to ensure that the
+  /// dirty state is thoroughly propagated as soon as any assets are changed.
+  Stream get onDirty => _onDirtyPool.stream;
+  final _onDirtyPool = new StreamPool.broadcast();
+
+  /// Whether this cluster is dirty and needs to be run.
+  bool get isDirty => _phases.any((phase) => phase.isDirty);
+
+  // TODO(nweiz): move to a more push-based way of propagating outputs and get
+  // rid of this.

[Bob Nystrom, 2013/10/04 21:47:50]

Add: "Once that's done, see if TransformerGroup and AssetCascade can be
unified."

[nweiz, 2013/10/07 23:21:31]

Done.

+  /// The set of outputs that has been returned by [process].
+  ///
+  /// [process] is expected to only return new outputs, so this is used to
+  /// ensure that it does so.
+  final _alreadyEmittedOutputs = new Set<AssetNode>();
+
+  ClusterRunner(AssetCascade cascade, TransformerCluster cluster) {
+    var lastPhase = new Phase(cascade, cluster.phases.first);

[Bob Nystrom, 2013/10/04 21:47:50]

It feels weird to me that the phases within the cluster know the top level
cascade that contains them. Maybe this requires unifying cluster and cascade,
but I wish a phase only knew the immediate thing that owned it.

[nweiz, 2013/10/07 23:21:31]

I find the fact that we're passing around the cascade at all weird. It only does
two things: provides access to cross-package assets via the PackageGraph, and
provides a channel for reporting error information. I'd rather handle the former
through some intermediate interface, and the latter by piping errors through
AssetNodes.

[Bob Nystrom, 2013/10/07 23:58:39]

SGTM. Maybe add a TODO?

[nweiz, 2013/10/08 00:30:39]

Done, in Phase.

+    _phases.add(lastPhase);
+    for (var phase in cluster.phases.skip(1)) {
+      lastPhase = lastPhase.addPhase(phase);
+      _phases.add(lastPhase);
+    }
+
+    for (var phase in _phases) {
+      _onDirtyPool.add(phase.onDirty);
+    }
+  }
+
+  /// Adds a new asset as an input for this cluster.
+  void addInput(AssetNode node) {
+    _phases.first.addInput(node);
+  }
+
+  /// Removes this cluster and all sub-phases within it.
+  void remove() {
+    _phases.first.remove();
+  }
+
+  /// Processes this cluster.
+  ///
+  /// Returns a future that completes with any new outputs produced by the
+  /// cluster.
+  Future<Set<AssetNode>> process() {
+    for (var phase in _phases) {

[Bob Nystrom, 2013/10/04 21:47:50]

Add comment, like "Process the first phase that needs to do work."

[nweiz, 2013/10/07 23:21:31]

Done.

+      var future = phase.process();
+      if (future != null) return future.then((_) => process());
+    }
+

[Bob Nystrom, 2013/10/04 21:47:50]

// If we get here, all phases are done processing.

[nweiz, 2013/10/07 23:21:31]

Done.

+    var newOutputs = _phases.last.availableOutputs
+        .difference(_alreadyEmittedOutputs);
+    for (var output in newOutputs) {
+      output.whenRemoved.then((_) => _alreadyEmittedOutputs.remove(output));
+    }
+    _alreadyEmittedOutputs.addAll(newOutputs);
+
+    return new Future.value(newOutputs);
+  }
+}