Move Barback to a more thoroughly push-based model.

pkg/barback/lib/src/group_runner.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.group_runner;
 
 import 'dart:async';
 
 import 'asset_cascade.dart';
 import 'asset_node.dart';
 import 'log.dart';
 import 'phase.dart';
 import 'stream_pool.dart';
 import 'transformer_group.dart';
 
 /// A class that processes all of the phases in a single transformer group.
 ///
 /// A group takes many inputs, processes them, and emits many outputs.
 class GroupRunner {
   /// The group this runner runs.
   final TransformerGroup _group;
 
   /// A string describing the location of [this] in the transformer graph.
   final String _location;
 
   /// The phases defined by this group.
   final _phases = new List<Phase>();
 
-  /// A stream that emits an event whenever this group becomes dirty and needs
-  /// to be run.
+  /// Whether [this] is dirty and still has more processing to do.
+  bool get isDirty => _phases.any((phase) => phase.isDirty);
+
+  /// A stream that emits an event whenever [this] is no longer dirty.
   ///
-  /// This may emit events when the group 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();
+  /// This is synchronous in order to guarantee that it will emit an event as
+  /// soon as [isDirty] flips from `true` to `false`.
+  Stream get onDone => _onDoneController.stream;
+  final _onDoneController = new StreamController.broadcast(sync: true);
 
-  /// Whether this group is dirty and needs to be run.
-  bool get isDirty => _phases.any((phase) => phase.isDirty);
+  /// A stream that emits any new assets emitted by [this].
+  ///
+  /// Assets are emitted synchronously to ensure that any changes are thoroughly
+  /// propagated as soon as they occur.
+  Stream<AssetNode> get onAsset => _onAssetPool.stream;
+  final _onAssetPool = new StreamPool<AssetNode>();
 
   /// A stream that emits an event whenever any transforms in this group logs
   /// an entry.
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
-  // TODO(nweiz): move to a more push-based way of propagating outputs and get
-  // rid of this. Once that's done, see if we can unify GroupRunner and
-  // AssetCascade.
-  /// 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>();
-
   GroupRunner(AssetCascade cascade, this._group, this._location) {
-    var lastPhase = new Phase(cascade, _group.phases.first, _location);
-    _phases.add(lastPhase);
+    _addPhase(new Phase(cascade, _location), _group.phases.first);
     for (var phase in _group.phases.skip(1)) {
-      lastPhase = lastPhase.addPhase(phase);
-      _phases.add(lastPhase);
-    }
-
-    for (var phase in _phases) {
-      _onDirtyPool.add(phase.onDirty);
-      _onLogPool.add(phase.onLog);
+      _addPhase(_phases.last.addPhase(), phase);
     }
   }
 
+  /// Add a phase with [contents] to [this]'s list of phases.
+  ///
+  /// [contents] should be an inner [Iterable] from a [TransformGroup.phases]
+  /// value.
+  void _addPhase(Phase phase, Iterable contents) {
+    _phases.add(phase);
+    _onAssetPool.add(phase.onAsset);
+    _onLogPool.add(phase.onLog);
+    phase.onDone.listen((_) {

[Bob Nystrom, 2014/03/05 22:13:25]

Do we need to listen to this on all phases, or just the last one?

[nweiz, 2014/03/06 00:29:08]

It's possible for a middle phase to go dirty and then back to done without
affecting anything else. For example, suppose the first phase had a transform
that just consumed an input and did nothing else. If that input changes, the
phase will go dirty until it finishes re-running [apply]. If it continues to do
nothing but consume the input, it will go clean again without affecting any
future phases.

Also, it's a pain to track which phase is the last one all the time, so even if
this wasn't relevant for other phases it would probably still be cleaner this
way.

+      if (!isDirty) _onDoneController.add(null);
+    });
+    phase.updateTransformers(contents);
+  }
+
   /// Force all [LazyTransformer]s' transforms in this group to begin producing
   /// concrete assets.
   void forceAllTransforms() {
     for (var phase in _phases) {
       phase.forceAllTransforms();
     }
   }
 
   /// Adds a new asset as an input for this group.
   void addInput(AssetNode node) {
     _phases.first.addInput(node);
   }
 
   /// Removes this group and all sub-phases within it.
   void remove() {
     _phases.first.remove();
   }
 
-  /// Processes this group.
-  ///
-  /// Returns a future that completes with any new outputs produced by the
-  /// group.
-  Future<Set<AssetNode>> process() {
-    // Process the first phase that needs to do work.
-    for (var phase in _phases) {
-      var future = phase.process();
-      if (future != null) return future.then((_) => process());
-    }
-
-    // If we get here, all phases are done processing.
-    var newOutputs = _phases.last.availableOutputs
-        .difference(_alreadyEmittedOutputs);
-    for (var output in newOutputs) {
-      output.whenRemoved(() => _alreadyEmittedOutputs.remove(output));
-    }
-    _alreadyEmittedOutputs.addAll(newOutputs);
-
-    return new Future.value(newOutputs);
-  }
-
   String toString() => "group in phase $_location for $_group";
 }