Add events in barback to bubble up the dirty bit.

pkg/barback/lib/src/phase.dart

Index: pkg/barback/lib/src/phase.dart
diff --git a/pkg/barback/lib/src/phase.dart b/pkg/barback/lib/src/phase.dart
index 4d2afb4dabcb9b929b904fe157cbf78789e7eb97..1fc82c0ff3a5f2ae7eae73b974c007c43bef7d0c 100644
--- a/pkg/barback/lib/src/phase.dart
+++ b/pkg/barback/lib/src/phase.dart
@@ -12,6 +12,7 @@ import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'errors.dart';
+import 'stream_pool.dart';
 import 'transform_node.dart';
 import 'transformer.dart';
 import 'utils.dart';
@@ -73,12 +74,31 @@ class Phase {
   /// output.
   final _outputs = new Set<AssetId>();
 
+  /// A stream that emits an event whenever this phase becomes dirty and needs
+  /// to be run.
+  ///
+  /// This may emit events when the phase 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();
+
+  /// A controller whose stream feeds into [_onDirtyPool].
+  ///
+  /// This is used whenever an input is added, changed, or removed. It's
+  /// sometimes redundant with the events collected from [_transforms], but this
+  /// stream is necessary for new and removed inputs, and the transform stream
+  /// is necessary for modified secondary inputs.
+  final _onDirtyController = new StreamController.broadcast(sync: true);
+
   /// The phase after this one.
   ///
   /// Outputs from this phase will be passed to it.
   final Phase _next;
 
-  Phase(this.cascade, this._index, this._transformers, this._next);
+  Phase(this.cascade, this._index, this._transformers, this._next) {
+    _onDirtyPool.addStream(_onDirtyController.stream);
+  }
 
   /// Adds a new asset as an input for this phase.
   ///
@@ -168,6 +188,11 @@ class Phase {
   /// time it takes to adjust its transformers, they're appropriately
   /// re-adjusted. Its progress can be tracked in [_adjustTransformersFutures].
   void _adjustTransformers(AssetNode node) {
+    // Mark the phase as dirty. This may not actually end up creating any new
+    // transforms, but we want adding or removing a source asset to consistently
+    // kick off a build, even if that build does nothing.

[Bob Nystrom, 2013/08/01 18:10:33]

Why?

[nweiz, 2013/08/01 20:30:52]

API consistency. If a user calls [Barback.updateSources] or
[Barback.removeSources] then waits for the build to complete, that should always
terminate. If we don't even kick off a build when we can prove nothing needs
building, we run the risk of deadlocking users who do that.

+    _onDirtyController.add(null);
+
     // Once the input is available, hook up transformers for it. If it changes
     // while that's happening, try again.
     _adjustTransformersFutures[node.id] = node.tryUntilStable((asset) {
@@ -182,7 +207,8 @@ class Phase {
       // future.
       node.onStateChange.first.then((state) {
         if (state.isRemoved) {
-          _transforms.remove(node.id);
+          _onDirtyController.add(null);
+          _removeTransforms(node.id);
         } else {
           _adjustTransformers(node);
         }
@@ -195,7 +221,7 @@ class Phase {
       // If the asset is removed, [tryUntilStable] will throw an
       // [AssetNotFoundException]. In that case, just remove all transforms for
       // the node.
-      _transforms.remove(node.id);
+      _removeTransforms(node.id);
     }).whenComplete(() {
       _adjustTransformersFutures.remove(node.id);
     });
@@ -205,6 +231,13 @@ class Phase {
     _adjustTransformersFutures[node.id].catchError((_) {});
   }
 
+  /// Remove all transforms for the asset identified by [id].
+  void _removeTransforms(AssetId id) {
+    for (var transform in _transforms.remove(id)) {
+      _onDirtyPool.removeStream(transform.onDirty);
+    }
+  }
+
   // Remove any old transforms that used to have [asset] as a primary asset but
   // no longer apply to its new contents.
   Future _removeStaleTransforms(Asset asset) {
@@ -214,6 +247,7 @@ class Phase {
       return transform.transformer.isPrimary(asset).then((isPrimary) {
         if (isPrimary) return;
         _transforms[asset.id].remove(transform);
+        _onDirtyPool.removeStream(transform.onDirty);
         transform.remove();
       });
     }));
@@ -240,7 +274,9 @@ class Phase {
       // results.
       return transformer.isPrimary(node.asset).then((isPrimary) {
         if (!isPrimary) return;
-        _transforms[node.id].add(new TransformNode(this, transformer, node));
+        var transform = new TransformNode(this, transformer, node);
+        _transforms[node.id].add(transform);
+        _onDirtyPool.addStream(transform.onDirty);
       });
     }));
   }