Move isPrimary computation from PhaseInput into TransformNode.

pkg/barback/lib/src/transform_node.dart

Index: pkg/barback/lib/src/transform_node.dart
diff --git a/pkg/barback/lib/src/transform_node.dart b/pkg/barback/lib/src/transform_node.dart
index 152ef1623c5ed18c9aaf9505b95edd389338d3f3..87cedae933cf06351580ecf69dc064b79b71128c 100644
--- a/pkg/barback/lib/src/transform_node.dart
+++ b/pkg/barback/lib/src/transform_node.dart
@@ -9,7 +9,6 @@ import 'dart:async';
 import 'asset.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
-import 'asset_set.dart';
 import 'declaring_transform.dart';
 import 'errors.dart';
 import 'lazy_transformer.dart';
@@ -41,26 +40,8 @@ class TransformNode {
   /// The subscription to [primary]'s [AssetNode.onStateChange] stream.
   StreamSubscription _primarySubscription;
 
-  // TODO(nweiz): Remove this and move isPrimary computation into TransformNode.
-  /// Whether the parent [PhaseInput] is currently computing whether its input
-  /// is primary for [this].
-  bool _pendingIsPrimary = false;
-
   /// Whether [this] is dirty and still has more processing to do.
-  bool get isDirty => _pendingIsPrimary || _isApplying;
-
-  /// Whether any input has become dirty since [_apply] last started running.
-  var _hasBecomeDirty = false;
-
-  /// Whether [_apply] is currently running.
-  var _isApplying = false;
-
-  /// Whether the most recent run of this transform has declared that it
-  /// consumes the primary input.
-  ///
-  /// Defaults to `false`.
-  bool get consumePrimary => _consumePrimary;
-  bool _consumePrimary = false;
+  bool get isDirty => !_state.isDone;
 
   /// Whether [transformer] is lazy and this transform has yet to be forced.
   bool _isLazy;
@@ -71,16 +52,21 @@ class TransformNode {
   /// The controllers for the asset nodes emitted by this node.
   var _outputControllers = new Map<AssetId, AssetNodeController>();
 
-  // TODO(nweiz): It's weird that this is different than the [onDone] stream the
-  // other nodes emit. See if we can make that more consistent.
-  /// A stream that emits an event whenever [onDirty] changes its value.
+  /// The controller that's used to pass [primary] through [this] if it's not
+  /// consumed or overwritten.
+  ///
+  /// This needs an intervening controller to ensure that the output can be
+  /// marked dirty when determining whether [this] will consume or overwrite it,
+  /// and be marked removed if it does. [_passThroughController] will be null
+  /// null if the asset is not being passed through.

[Bob Nystrom, 2014/03/12 17:42:58]

"null null" -> "null"

[nweiz, 2014/03/12 21:44:47]

Done.

+  AssetNodeController _passThroughController;
+
+  /// 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] changes. It's possible for this to emit multiple events
-  /// while [isDirty] is `true`. However, it will only emit a single event each
-  /// time [isDirty] becomes `false`.
-  Stream get onStateChange => _onStateChangeController.stream;
-  final _onStateChangeController = new StreamController.broadcast(sync: true);
+  /// soon as [isDirty] flips from `true` to `false`.
+  Stream get onDone => _onDoneController.stream;
+  final _onDoneController = new StreamController.broadcast(sync: true);
 
   /// A stream that emits any new assets emitted by [this].
   ///
@@ -97,6 +83,19 @@ class TransformNode {
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
+  /// The current state of [this].
+  var _state = _TransformNodeState.PROCESSING;
+
+  /// Whether [this] has been marked as removed.
+  bool get _isRemoved => _onAssetController.isClosed;
+
+  /// Whether the most recent run of this transform has declared that it
+  /// consumes the primary input.
+  ///
+  /// Defaults to `false`. This is not meaningful unless [_state] is
+  /// [_TransformNodeState.APPLIED].
+  bool _consumePrimary = false;
+
   TransformNode(this.phase, Transformer transformer, this.primary,
       this._location)
       : transformer = transformer,
@@ -105,12 +104,11 @@ class TransformNode {
       if (state.isRemoved) {
         remove();
       } else {
-        if (state.isDirty) _pendingIsPrimary = true;
-        _dirty();
+        _dirty(primaryChanged: true);
       }
     });
 
-    _apply();
+    _process();
   }
 
   /// The [TransformInfo] describing this node.
@@ -126,16 +124,17 @@ class TransformNode {
   /// from the primary input, but it's possible for a transform to no longer be
   /// valid even if its primary input still exists.
   void remove() {
-    _hasBecomeDirty = false;
     _onAssetController.close();
-    _onStateChangeController.close();
+    _onDoneController.close();
     _primarySubscription.cancel();
-    for (var subscription in _inputSubscriptions.values) {
-      subscription.cancel();
-    }
+    _clearInputSubscriptions();
     for (var controller in _outputControllers.values) {
       controller.setRemoved();
     }
+    if (_passThroughController != null) {
+      _passThroughController.setRemoved();
+      _passThroughController = null;
+    }
   }
 
   /// If [transformer] is lazy, ensures that its concrete outputs will be
@@ -145,47 +144,83 @@ class TransformNode {
     // transform's outputs have gone unused, we switch it back to lazy mode.
     if (!_isLazy) return;
     _isLazy = false;
-    _dirty();
-  }
-
-  // TODO(nweiz): remove this and move isPrimary computation into TransformNode.
-  /// Mark that the parent [PhaseInput] has determined that its input is indeed
-  /// primary for [this].
-  void markPrimary() {
-    if (!_pendingIsPrimary) return;
-    _pendingIsPrimary = false;
-    if (!_isApplying) _apply();
+    _dirty(primaryChanged: false);
   }
 
   /// Marks this transform as dirty.
   ///
   /// This causes all of the transform's outputs to be marked as dirty as well.
-  void _dirty() {
+  /// [primaryChanged] should be true if and only if [this] was set dirty
+  /// because [primary] changed.
+  void _dirty({bool primaryChanged: false}) {
+    if (!primaryChanged && _state.isNotPrimary) return;
+
+    if (_passThroughController != null) _passThroughController.setDirty();
     for (var controller in _outputControllers.values) {
       controller.setDirty();
     }
 
-    _hasBecomeDirty = true;
-    _onStateChangeController.add(null);
-    if (!_isApplying && !_pendingIsPrimary) _apply();
+    if (_state.isDone) {
+      if (primaryChanged) {
+        _process();
+      } else {
+        _apply();
+      }
+    } else if (primaryChanged) {
+      _state = _TransformNodeState.NEEDS_IS_PRIMARY;
+    } else if (!_state.needsIsPrimary) {
+      _state = _TransformNodeState.NEEDS_APPLY;
+    }

[Bob Nystrom, 2014/03/12 17:42:58]

It might be worth moving this conditional logic into the state itself:

_state = _state._dirty(primaryChanged: primaryChanged);

[nweiz, 2014/03/12 21:44:47]

I tried this out, but I ended up finding it more confusing than keeping all the
logic in TransformNode. The interaction between setting the state and doing a
bunch of other actions to the transform node got more twisted when I moved some
of them out into the state class.

+  }
+
+  /// Determines whether [primary] is primary for [transformer], and if so runs
+  /// [transformer.apply].
+  void _process() {
+    // Clear all the old input subscriptions. If an input is re-used, we'll
+    // re-subscribe.
+    _clearInputSubscriptions();
+    _state = _TransformNodeState.PROCESSING;
+    primary.whenAvailable((_) {
+      _state = _TransformNodeState.PROCESSING;
+      return transformer.isPrimary(primary.asset);
+    }).catchError((error, stackTrace) {
+      // If the transform became dirty while processing, ignore any errors from
+      // it.
+      if (_state.needsIsPrimary || _isRemoved) return false;
+
+      if (error is! MissingInputException) {
+        error = new TransformerException(info, error, stackTrace);
+      }
+
+      // Catch all transformer errors and pipe them to the results stream. This
+      // is so a broken transformer doesn't take down the whole graph.
+      phase.cascade.reportError(error);
+
+      return false;
+    }).then((isPrimary) {
+      if (_isRemoved) return;
+      if (_state.needsIsPrimary) {
+        _process();
+      } else if (isPrimary) {
+        _apply();
+      } else {
+        _doesNotApply();
+        _state = _TransformNodeState.NOT_PRIMARY;
+        _onDoneController.add(null);
+      }

[Bob Nystrom, 2014/03/12 17:42:58]

This might be another chunk of conditional logic you could push into the state:

_state = _state.finishIsPrimary(this, isPrimary);

+    });
   }
 
   /// Applies this transform.
   void _apply() {
     assert(!_onAssetController.isClosed);
 
-    // Clear all the old input subscriptions. If an input is re-used, we'll
-    // re-subscribe.
-    for (var subscription in _inputSubscriptions.values) {
-      subscription.cancel();
-    }
-    _inputSubscriptions.clear();
-
-    _isApplying = true;
-    _onStateChangeController.add(null);
+    // Clear input subscriptions here as well as in [_process] because [_apply]
+    // may be restarted independently if only a secondary input changes.
+    _clearInputSubscriptions();
+    _state = _TransformNodeState.PROCESSING;
     primary.whenAvailable((_) {
-      _hasBecomeDirty = false;
-
+      if (_state.needsApply) _state = _TransformNodeState.PROCESSING;
       // TODO(nweiz): If [transformer] is a [DeclaringTransformer] but not a
       // [LazyTransformer], we can get some mileage out of doing a declarative
       // first so we know how to hook up the assets.
@@ -194,7 +229,7 @@ class TransformNode {
     }).catchError((error, stackTrace) {
       // If the transform became dirty while processing, ignore any errors from
       // it.
-      if (_hasBecomeDirty || _onAssetController.isClosed) return;
+      if (!_state.isProcessing || _isRemoved) return;
 
       if (error is! MissingInputException) {
         error = new TransformerException(info, error, stackTrace);
@@ -204,22 +239,18 @@ class TransformNode {
       // is so a broken transformer doesn't take down the whole graph.
       phase.cascade.reportError(error);
 
-      // Remove all the previously-emitted assets.
-      for (var controller in _outputControllers.values) {
-        controller.setRemoved();
-      }
-      _outputControllers.clear();
+      _doesNotApply();
     }).then((_) {
-      if (_onAssetController.isClosed) return;
+      if (_isRemoved) return;
 
-      _isApplying = false;
-      if (_hasBecomeDirty) {
-        // Re-apply the transform if it became dirty while applying.
-        if (!_pendingIsPrimary) _apply();
+      if (_state.needsIsPrimary) {
+        _process();
+      } else if (_state.needsApply) {
+        _apply();
       } else {
-        assert(!isDirty);
-        // Otherwise, notify the parent nodes that it's no longer dirty.
-        _onStateChangeController.add(null);
+        assert(_state.isProcessing);
+        _state = _TransformNodeState.APPLIED;
+        _onDoneController.add(null);
       }

[Bob Nystrom, 2014/03/12 17:42:58]

Ditto.

     });
   }
@@ -235,8 +266,9 @@ class TransformNode {
       // results stream.
       if (node == null) throw new MissingInputException(info, id);
 
-      _inputSubscriptions.putIfAbsent(node.id,
-          () => node.onStateChange.listen((_) => _dirty()));
+      _inputSubscriptions.putIfAbsent(node.id, () {
+        return node.onStateChange.listen((_) => _dirty(primaryChanged: false));
+      });
 
       return node.asset;
     });
@@ -251,7 +283,7 @@ class TransformNode {
     return syncFuture(() {
       return transformer.apply(transformController.transform);
     }).then((_) {
-      if (_hasBecomeDirty || _onAssetController.isClosed) return;
+      if (!_state.isProcessing || _onAssetController.isClosed) return;
 
       _consumePrimary = transformController.consumePrimary;
 
@@ -273,6 +305,14 @@ class TransformNode {
         _outputControllers.remove(id).setRemoved();
       }
 
+      // Emit or stop emitting the pass-through asset between removing and
+      // adding outputs to ensure there are no collisions.
+      if (!newOutputs.containsId(primary.id)) {
+        _emitPassThrough();
+      } else {
+        _dontEmitPassThrough();
+      }
+
       // Store any new outputs or new contents for existing outputs.
       for (var asset in newOutputs) {
         var controller = _outputControllers[asset.id];
@@ -296,7 +336,7 @@ class TransformNode {
       return (transformer as LazyTransformer)
           .declareOutputs(transformController.transform);
     }).then((_) {
-      if (_hasBecomeDirty || _onAssetController.isClosed) return;
+      if (!_state.isProcessing || _onAssetController.isClosed) return;
 
       _consumePrimary = transformController.consumePrimary;
 
@@ -315,6 +355,14 @@ class TransformNode {
         _outputControllers.remove(id).setRemoved();
       }
 
+      // Emit or stop emitting the pass-through asset between removing and
+      // adding outputs to ensure there are no collisions.
+      if (!newIds.contains(primary.id)) {
+        _emitPassThrough();
+      } else {
+        _dontEmitPassThrough();
+      }
+
       for (var id in newIds) {
         var controller = _outputControllers[id];
         if (controller != null) {
@@ -328,6 +376,116 @@ class TransformNode {
     });
   }
 
+  /// Cancels all subscriptions to secondary input nodes.
+  void _clearInputSubscriptions() {
+    for (var subscription in _inputSubscriptions.values) {
+      subscription.cancel();
+    }
+    _inputSubscriptions.clear();
+  }
+
+  /// Marks this transformer as not applying to [primary].
+  ///
+  /// This might be because [primary] isn't primary for [transformer], or
+  /// because [transformer] threw an error during [transformer.apply].
+  void _doesNotApply() {
+    // Remove all the previously-emitted assets.
+    for (var controller in _outputControllers.values) {
+      controller.setRemoved();
+    }
+    _outputControllers.clear();
+    _emitPassThrough();
+  }
+
+  /// Emit the pass-through asset if it's not being emitted already.
+  void _emitPassThrough() {
+    assert(!_outputControllers.containsKey(primary.id));
+
+    if (_consumePrimary) return;
+    if (_passThroughController == null) {
+      _passThroughController = new AssetNodeController.from(primary);
+      _onAssetController.add(_passThroughController.node);
+    } else {
+      _passThroughController.setAvailable(primary.asset);
+    }
+  }
+
+  /// Stop emitting the pass-through asset if it's being emitted already.
+  void _dontEmitPassThrough() {
+    if (_passThroughController == null) return;
+    _passThroughController.setRemoved();
+    _passThroughController = null;
+  }
+
   String toString() =>
     "transform node in $_location for $transformer on $primary";
 }
+
+/// The enum of states that [TransformNode] can be in.
+class _TransformNodeState {
+  /// The transform node is running [Transformer.isPrimary] or
+  /// [Transformer.apply] and doesn't need to re-run them.
+  ///
+  /// If there are no external changes by the time the processing finishes, this
+  /// will transition to [APPLIED] or [NOT_PRIMARY] depending on the result of
+  /// [Transformer.isPrimary]. If the primary input changes, this will
+  /// transition to [NEEDS_IS_PRIMARY]. If a secondary input changes, this will
+  /// transition to [NEEDS_APPLY].
+  static final PROCESSING = const _TransformNodeState._("processing");
+
+  /// The transform is running [Transformer.isPrimary] or [Transformer.apply],
+  /// but since it started the primary input changed, so it will need to re-run
+  /// [Transformer.isPrimary].
+  ///
+  /// This will always transition to [Transformer.PROCESSING].
+  static final NEEDS_IS_PRIMARY =
+    const _TransformNodeState._("needs isPrimary");
+
+  /// The transform is running [Transformer.apply], but since it started a
+  /// secondary input changed, so it will need to re-run [Transformer.apply].
+  ///
+  /// If there are no external changes by the time [Transformer.apply] finishes,
+  /// this will transition to [PROCESSING]. If the primary input changes, this
+  /// will transition to [NEEDS_IS_PRIMARY].
+  static final NEEDS_APPLY = const _TransformNodeState._("needs apply");
+
+  /// The transform has finished running [Transformer.apply], whether or not it
+  /// emitted an error.
+  ///
+  /// If the primary input or a secondary input changes, this will transition to
+  /// [PROCESSING].
+  static final APPLIED = const _TransformNodeState._("applied");
+
+  /// The transform has finished running [Transformer.isPrimary], which returned
+  /// `false`.
+  ///
+  /// If the primary input changes, this will transition to [PROCESSING].
+  static final NOT_PRIMARY = const _TransformNodeState._("not primary");
+
+  /// Whether [this] is [PROCESSING].
+  bool get isProcessing => this == _TransformNodeState.PROCESSING;
+
+  /// Whether [this] is [NEEDS_IS_PRIMARY].
+  bool get needsIsPrimary => this == _TransformNodeState.NEEDS_IS_PRIMARY;
+
+  /// Whether [this] is [NEEDS_APPLY].
+  bool get needsApply => this == _TransformNodeState.NEEDS_APPLY;
+
+  /// Whether [this] is [APPLIED].
+  bool get isApplied => this == _TransformNodeState.APPLIED;
+
+  /// Whether [this] is [IS_NOT_PRIMARY].

[Bob Nystrom, 2014/03/12 17:42:58]

IS_NOT_PRIMARY -> NOT_PRIMARY.

[nweiz, 2014/03/12 21:44:47]

Done.

+  bool get isNotPrimary => this == _TransformNodeState.NOT_PRIMARY;
+
+  /// Whether the transform has finished running [Transformer.isPrimary] and
+  /// [Transformer.apply].
+  ///
+  /// Specifically, whether [this] is [APPLIED] or [IS_NOT_PRIMARY].
+  bool get isDone => isApplied || isNotPrimary;
+
+  final String name;
+
+  const _TransformNodeState._(this.name);
+
+  String toString() => name;
+}