Move Barback to a more thoroughly push-based model.

pkg/barback/lib/src/phase_input.dart

Index: pkg/barback/lib/src/phase_input.dart
diff --git a/pkg/barback/lib/src/phase_input.dart b/pkg/barback/lib/src/phase_input.dart
index d89aedaab6fc8454b42c6a0c6fe7cbb5faf17c50..3cf1f0bd1e12265d60957d472b539c171e9fc1c5 100644
--- a/pkg/barback/lib/src/phase_input.dart
+++ b/pkg/barback/lib/src/phase_input.dart
@@ -5,9 +5,7 @@
 library barback.phase_input;
 
 import 'dart:async';
-import 'dart:collection';
 
-import 'asset.dart';
 import 'asset_forwarder.dart';
 import 'asset_node.dart';
 import 'errors.dart';
@@ -53,34 +51,39 @@ class PhaseInput {
   /// they do. It's null if the asset is not being passed through.
   AssetNodeController _passThroughController;
 
-  /// Whether [_passThroughController] has been newly created since [process]
-  /// last completed.
-  bool _newPassThrough = false;
-
-  /// A Future that will complete once the transformers that consume [input] are
-  /// determined.
-  Future _adjustTransformersFuture;
+  /// A stream that emits an event whenever [this] is no longer dirty.
+  ///
+  /// 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);
 
-  /// A stream that emits an event whenever this input becomes dirty and needs
-  /// [process] to be called.
+  /// A stream that emits any new assets emitted by [this].
   ///
-  /// This may emit events when the input 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();
+  /// 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 controller whose stream feeds into [_onDirtyPool].
+  /// A controller for emitting assets.
   ///
-  /// This is used whenever the input is changed or removed. It's sometimes
-  /// redundant with the events collected from [_transforms], but this stream is
-  /// necessary for removed inputs, and the transform stream is necessary for
-  /// modified secondary inputs.
-  final _onDirtyController = new StreamController.broadcast(sync: true);
+  /// This will be added to [_onAssetPool]. It's used to emit pass-through
+  /// assets.
+  final _onAssetController = new StreamController<AssetNode>(sync: true);
+
+  /// Whether [this] is dirty and still has more processing to do.
+  bool get isDirty => _isAdjustingTransformers ||
+      _transforms.any((transform) => transform.isDirty);
 
-  /// Whether this input is dirty and needs [process] to be called.
-  bool get isDirty => _adjustTransformersFuture != null ||
-      _newPassThrough || _transforms.any((transform) => transform.isDirty);
+  /// Whether [this] has been rmeoved.
+  bool get _isRemoved => _onAssetController.isClosed;
+
+  /// Whether [input] has become dirty since [_adjustTransformers] last started
+  /// running.
+  bool _hasBecomeDirty = false;
+
+  /// Whether [_isAdjustingTransformers] is currently running.
+  bool _isAdjustingTransformers = false;
 
   /// A stream that emits an event whenever any transforms that use [input] as
   /// their primary input log an entry.
@@ -91,13 +94,13 @@ class PhaseInput {
       this._location)
       : _transformers = transformers.toSet(),
         _inputForwarder = new AssetForwarder(input) {
-    _onDirtyPool.add(_onDirtyController.stream);
+    _onAssetPool.add(_onAssetController.stream);
 
     input.onStateChange.listen((state) {
       if (state.isRemoved) {
         remove();
-      } else if (_adjustTransformersFuture == null) {
-        _adjustTransformers();
+      } else {
+        _dirty();
       }
     });
 
@@ -108,8 +111,10 @@ class PhaseInput {
   ///
   /// This marks all outputs of the input as removed.
   void remove() {
-    _onDirtyController.add(null);
-    _onDirtyPool.close();
+    _onDoneController.close();
+    _hasBecomeDirty = false;
+    _onAssetPool.close();
+    _onAssetController.close();
     _onLogPool.close();
     _inputForwarder.close();
     if (_passThroughController != null) {
@@ -118,36 +123,31 @@ class PhaseInput {
     }
   }
 
+  /// Mark [this] as dirty and start re-running [_adjustTransformers] if
+  /// necessary.
+  void _dirty() {
+    // If there's a pass-through for this input, mark it dirty until we figure
+    // out whether we need to add any transforms for it.
+    if (_passThroughController != null) _passThroughController.setDirty();
+    _hasBecomeDirty = true;
+    if (!_isAdjustingTransformers) _adjustTransformers();
+  }
+
   /// Set this input's transformers to [transformers].
   void updateTransformers(Iterable<Transformer> newTransformersIterable) {
     var newTransformers = newTransformersIterable.toSet();
     var oldTransformers = _transformers.toSet();
-    for (var removedTransformer in
-         oldTransformers.difference(newTransformers)) {
+    var removedTransformers = oldTransformers.difference(newTransformers);
+    for (var removedTransformer in removedTransformers) {
       _transformers.remove(removedTransformer);
-
-      // If the transformers are being adjusted for [id], it will
-      // automatically pick up on [removedTransformer] being gone.
-      if (_adjustTransformersFuture != null) continue;
-
-      _transforms.removeWhere((transform) {
-        if (transform.transformer != removedTransformer) return false;
-        transform.remove();
-        return true;
-      });
-    }
-
-    if (_transforms.isEmpty && _adjustTransformersFuture == null &&
-        _passThroughController == null) {
-      _passThroughController = new AssetNodeController.from(input);
-      _newPassThrough = true;
     }
 
     var brandNewTransformers = newTransformers.difference(oldTransformers);
-    if (brandNewTransformers.isEmpty) return;
-
     brandNewTransformers.forEach(_transformers.add);
-    if (_adjustTransformersFuture == null) _adjustTransformers();
+
+    if (removedTransformers.isNotEmpty || brandNewTransformers.isNotEmpty) {
+      _dirty();
+    }
   }
 
   /// Force all [LazyTransformer]s' transforms in this input to begin producing
@@ -163,52 +163,65 @@ class PhaseInput {
   ///
   /// This ensures that if [input] 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 [_adjustTransformersFuture].
+  /// re-adjusted.
   void _adjustTransformers() {
-    // Mark the input 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.
-    _onDirtyController.add(null);
+    assert(!_isRemoved);
 
-    // If there's a pass-through for this input, mark it dirty while we figure
-    // out whether we need to add any transforms for it.
-    if (_passThroughController != null) _passThroughController.setDirty();
+    _isAdjustingTransformers = true;
+    input.whenAvailable((asset) {
+      _hasBecomeDirty = false;
 
-    // Once the input is available, hook up transformers for it. If it changes
-    // while that's happening, try again.
-    _adjustTransformersFuture = _tryUntilStable((asset, transformers) {
+      // Take a snapshot of the existing transformers that apply to this input.
+      // Since [_removeStaleTransforms] will check each of these transformers to
+      // be sure [input] is still primary for them, we use this set to avoid
+      // needlessly re-checking in [_addFreshTransforms].
       var oldTransformers =
           _transforms.map((transform) => transform.transformer).toSet();
 
-      return _removeStaleTransforms(asset, transformers).then((_) =>
-          _addFreshTransforms(transformers, oldTransformers));
-    }).then((_) => _adjustPassThrough()).catchError((error) {
-      if (error is! AssetNotFoundException || error.id != input.id) {
-        throw error;
-      }
+      return _removeStaleTransforms().then((_) {
+        if (_hasBecomeDirty || _isRemoved) return null;
+        return _addFreshTransforms(oldTransformers);
+      });
+    }).then((_) {
+      if (_hasBecomeDirty || _isRemoved) return null;
+      _adjustPassThrough();
+    }).catchError((error, stackTrace) {
+      if (error is! AssetNotFoundException || error.id != input.id) throw error;
 
-      // If the asset is removed, [_tryUntilStable] will throw an
+      // If the asset is removed, [input.whenAvailable] will throw an
       // [AssetNotFoundException]. In that case, just remove it.
       remove();
-    }).whenComplete(() {
-      _adjustTransformersFuture = null;
+    }).then((_) {
+      if (_isRemoved) return;
+
+      _isAdjustingTransformers = false;
+      if (_hasBecomeDirty) {
+        _adjustTransformers();
+      } else if (!isDirty) {
+        _onDoneController.add(null);
+      }
     });
   }
 
-  // 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, Set<Transformer> transformers) {
+  // Remove any old transforms that used to have [input]'s asset as a primary
+  // asset but no longer apply to its new contents.
+  Future _removeStaleTransforms() {
+    assert(input.state.isAvailable);
+
     return Future.wait(_transforms.map((transform) {
-      return newFuture(() {
-        if (!transformers.contains(transform.transformer)) return false;
+      return syncFuture(() {
+        if (!_transformers.contains(transform.transformer)) return false;
 
         // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
         // results (issue 16162).
-        return transform.transformer.isPrimary(asset);
+        return transform.transformer.isPrimary(input.asset);
       }).then((isPrimary) {
-        if (isPrimary) return;
-        _transforms.remove(transform);
-        transform.remove();
+        if (_hasBecomeDirty) return;
+        if (isPrimary) {
+          transform.markPrimary();
+        } else if (_transforms.remove(transform)) {
+          transform.remove();
+        }
       });
     }));
   }
@@ -220,27 +233,24 @@ class PhaseInput {
   // transforms that had [input] as a primary input prior to this. They don't
   // need to be checked, since their transforms were removed or preserved in
   // [_removeStaleTransforms].
-  Future _addFreshTransforms(Set<Transformer> transformers,
-      Set<Transformer> oldTransformers) {
-    return Future.wait(transformers.map((transformer) {
-      if (oldTransformers.contains(transformer)) return new Future.value();
+  Future _addFreshTransforms(Set<Transformer> oldTransformers) {
+    assert(input.state.isAvailable);
 
-      // If the asset is unavailable, the results of this [_adjustTransformers]
-      // run will be discarded, so we can just short-circuit.
-      if (input.asset == null) return new Future.value();
+    return Future.wait(_transformers.map((transformer) {
+      if (oldTransformers.contains(transformer)) return new Future.value();
 
-      // We can safely access [input.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(input.asset).then((isPrimary) {
-        if (!isPrimary) return;
+        if (_hasBecomeDirty || !isPrimary) return;
         var transform = new TransformNode(
             _phase, transformer, input, _location);
         _transforms.add(transform);
-        _onDirtyPool.add(transform.onDirty);
+        _onAssetPool.add(transform.onAsset);
         _onLogPool.add(transform.onLog);
+        transform.onDone.listen((_) {
+          if (!isDirty) _onDoneController.add(null);
+        }, onDone: () => _transforms.remove(transform));
       });
     }));
   }
@@ -258,68 +268,12 @@ class PhaseInput {
         _passThroughController.setAvailable(input.asset);
       } else {
         _passThroughController = new AssetNodeController.from(input);
-        _newPassThrough = true;
+        _onAssetController.add(_passThroughController.node);
       }
     } else if (_passThroughController != null) {
       _passThroughController.setRemoved();
       _passThroughController = null;
-      _newPassThrough = false;
-    }
-  }
-
-  /// Like [AssetNode.tryUntilStable], but also re-runs [callback] if this
-  /// phase's transformers are modified.
-  Future _tryUntilStable(
-      Future callback(Asset asset, Set<Transformer> transformers)) {
-    var oldTransformers;
-    return input.tryUntilStable((asset) {
-      oldTransformers = _transformers.toSet();
-      return callback(asset, _transformers);
-    }).then((result) {
-      if (setEquals(oldTransformers, _transformers)) return result;
-      return _tryUntilStable(callback);
-    });
-  }
-
-  /// Processes the transforms for this input.
-  ///
-  /// Returns the set of newly-created asset nodes that transforms have emitted
-  /// for this input. The assets returned this way are guaranteed not to be
-  /// [AssetState.REMOVED].
-  Future<Set<AssetNode>> process() {
-    return _waitForTransformers(() => _processTransforms()).then((outputs) {
-      if (input.state.isRemoved) return new Set();
-      return outputs;
-    });
-  }
-
-  /// Runs [callback] once all the transformers are adjusted correctly and the
-  /// input is ready to be processed.
-  ///
-  /// If the transformers are already properly adjusted, [callback] is called
-  /// synchronously to ensure that [_adjustTransformers] isn't called before the
-  /// callback.
-  Future _waitForTransformers(callback()) {
-    if (_adjustTransformersFuture == null) return syncFuture(callback);
-    return _adjustTransformersFuture.then(
-        (_) => _waitForTransformers(callback));
-  }
-
-  /// Applies all currently wired up and dirty transforms.
-  Future<Set<AssetNode>> _processTransforms() {
-    if (input.state.isRemoved) return new Future.value(new Set());
-
-    if (_passThroughController != null) {
-      if (!_newPassThrough) return new Future.value(new Set());
-      _newPassThrough = false;
-      return new Future.value(
-          new Set<AssetNode>.from([_passThroughController.node]));
     }
-
-    return Future.wait(_transforms.map((transform) {
-      if (!transform.isDirty) return new Future.value(new Set());
-      return transform.apply();
-    })).then((outputs) => unionAll(outputs));
   }
 
   String toString() => "phase input in $_location for $input";