Add events in barback to bubble up the dirty bit.

pkg/barback/lib/src/phase.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.phase;
 
 import 'dart:async';
 
 import 'asset.dart';
 import 'asset_cascade.dart';
 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';
 
 /// One phase in the ordered series of transformations in an [AssetCascade].
 ///
 /// Each phase can access outputs from previous phases and can in turn pass
 /// outputs to later phases. Phases are processed strictly serially. All
 /// transforms in a phase will be complete before moving on to the next phase.
 /// Within a single phase, all transforms will be run in parallel.
@@ skipping 41 matching lines @@
   /// New asset nodes that were added while [_adjustTransformers] was still
   /// being run on an old version of that asset.
   var _pendingNewInputs = new Map<AssetId, AssetNode>();
 
   /// The ids of assets that are emitted by transforms in this phase.
   ///
   /// This is used to detect collisions where multiple transforms emit the same
   /// 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.
   ///
   /// [node] doesn't have to be [AssetState.AVAILABLE]. Once it is, the phase
   /// will automatically begin determining which transforms can consume it as a
   /// primary input. The transforms themselves won't be applied until [process]
   /// is called, however.
   ///
   /// This should only be used for brand-new assets or assets that have been
   /// removed and re-created. The phase will automatically handle updated assets
@@ skipping 69 matching lines @@
     return inputs[id];
   }
 
   /// Asynchronously determines which transformers can consume [node] as a
   /// primary input and creates transforms for them.
   ///
   /// This ensures that if [node] is modified or removed during or after the
   /// 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) {
       var oldTransformers = _transforms[node.id]
           .map((transform) => transform.transformer).toSet();
 
       return _removeStaleTransforms(asset)
           .then((_) => _addFreshTransforms(node, oldTransformers));
     }).then((_) {
       // Now all the transforms are set up correctly and the asset is available
       // for the time being. Set up handlers for when the asset changes in the
       // future.
       node.onStateChange.first.then((state) {
         if (state.isRemoved) {
-          _transforms.remove(node.id);
+          _onDirtyController.add(null);
+          _removeTransforms(node.id);
         } else {
           _adjustTransformers(node);
         }
       }).catchError((e) {
         _adjustTransformersFutures[node.id] = new Future.error(e);
       });
     }).catchError((error) {
       if (error is! AssetNotFoundException || error.id != node.id) throw error;
 
       // 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);
     });
 
     // Don't top-level errors coming from the input processing. Any errors will
     // eventually be piped through [process]'s returned Future.
     _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) {
     return Future.wait(_transforms[asset.id].map((transform) {
       // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
       // results.
       return transform.transformer.isPrimary(asset).then((isPrimary) {
         if (isPrimary) return;
         _transforms[asset.id].remove(transform);
+        _onDirtyPool.removeStream(transform.onDirty);
         transform.remove();
       });
     }));
   }
 
   // Add new transforms for transformers that consider [node]'s asset to be a
   // primary input.
   //
   // [oldTransformers] is the set of transformers that had [node] as a primary
   // input prior to this. They don't need to be checked, since they were removed
   // or preserved in [_removeStaleTransforms].
   Future _addFreshTransforms(AssetNode node, Set<Transformer> oldTransformers) {
     return Future.wait(_transformers.map((transformer) {
       if (oldTransformers.contains(transformer)) return new Future.value();
 
       // If the asset is unavailable, the results of this [_adjustTransformers]
       // run will be discarded, so we can just short-circuit.
       if (node.asset == null) return new Future.value();
 
       // We can safely access [node.asset] here even though it might have
       // changed since (as above) if it has, [_adjustTransformers] will just be
       // re-run.
       // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
       // 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);
       });
     }));
   }
 
   /// Processes this phase.
   ///
   /// Returns a future that completes when processing is done. If there is
   /// nothing to process, returns `null`.
   Future process() {
     if (_adjustTransformersFutures.isEmpty) return _processTransforms();
@@ skipping 33 matching lines @@
       // Report collisions in a deterministic order.
       collisions = collisions.toList();
       collisions.sort((a, b) => a.toString().compareTo(b.toString()));
       for (var collision in collisions) {
         cascade.reportError(new AssetCollisionException(collision));
         // TODO(rnystrom): Define what happens after a collision occurs.
       }
     });
   }
 }