Make Phase.getInput in barback play nicely with the push model.

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_cascade.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
+import 'errors.dart';
 import 'group_runner.dart';
 import 'log.dart';
 import 'multiset.dart';
 import 'phase_forwarder.dart';
 import 'phase_input.dart';
 import 'phase_output.dart';
 import 'stream_pool.dart';
 import 'transformer.dart';
 import 'transformer_group.dart';
 import 'utils.dart';
@@ skipping 67 matching lines @@
   /// Assets are emitted synchronously to ensure that any changes are thoroughly
   /// propagated as soon as they occur. Only a phase with no [next] phase will
   /// emit assets.
   Stream<AssetNode> get onAsset => _onAssetController.stream;
   final _onAssetController = new StreamController<AssetNode>(sync: true);
 
   /// Whether [this] is dirty and still has more processing to do.
   bool get isDirty => _inputs.values.any((input) => input.isDirty) ||
       _groups.values.any((group) => group.isDirty);
 
+  /// Whether [this] or any previous phase is dirty.
+  bool get _isTransitivelyDirty => isDirty ||
+      (_previous != null && _previous._isTransitivelyDirty);
+
   /// A stream that emits an event whenever any transforms in this phase logs
   /// an entry.
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
+  /// The previous phase in the cascade, or null if this is the first phase.
+  final Phase _previous;
+
   /// The phase after this one.
   ///
   /// Outputs from this phase will be passed to it.
   Phase get next => _next;
   Phase _next;
 
+  /// A map of asset ids to completers for [getInput] requests.
+  ///
+  /// If an asset node is requested before it's available, we put a completer in
+  /// this map to wait for the asset to be generated. If it's not generated, the
+  /// completer should complete to `null`.
+  final _pendingOutputRequests = new Map<AssetId, Completer<AssetNode>>();
+
   /// Returns all currently-available output assets for this phase.
   Set<AssetNode> get availableOutputs {
     return _outputs.values
         .map((output) => output.output)
         .where((node) => node.state.isAvailable)
         .toSet();
   }
 
   // TODO(nweiz): Rather than passing the cascade and the phase everywhere,
   // create an interface that just exposes [getInput]. Emit errors via
   // [AssetNode]s.
   Phase(AssetCascade cascade, String location)
       : this._(cascade, location, 0);
 
-  Phase._(this.cascade, this._location, this._index);
+  Phase._(this.cascade, this._location, this._index, [this._previous]) {
+    for (var phase = this; phase != null; phase = phase._previous) {
+      phase.onDone.listen((_) {
+        if (_isTransitivelyDirty) return;

[Bob Nystrom, 2014/03/06 23:59:22]

If we're listening on every phase, why do we need to check for transitive
dirtiness?

This is, what, n^2? Can you do this a more efficient way?

[nweiz, 2014/03/07 00:20:19]

This listener only fires when one phase finishes its work, but each phase's
dirty state is independent of the others'. We can't guarantee that an output
won't be generated until *no* parent phases are dirty, which is what this
checks.
Which part? [_isTransitivelyDirty] is O(n) in the number of phases, since it
just walks upward through the phase tree. Creating a bunch of phases is O(n^2),
since each one has to create a listener for all the phases above, but that only
happens once when the cascade is created. We could theoretically make that O(n)
as well by adding an [onTransitivelyDone] stream, but that feels like too much
extra API surface for too little gain.

[Bob Nystrom, 2014/03/07 00:54:15]

Say there are phases A, B, C, and D. A's onDone fires:

B receives it and checks to see if A is done.
C receives it and checks to see if A and B are done.
D receives it and checks to see if A, B, and C are done.
That feels a bit gross too, but something that only requires looking at the
previous phase seems like a better solution here. Given how hard you try to keep
things encapsulated, it seems weird that here you're looping over all of the
previous phases.

[nweiz, 2014/03/07 01:08:27]

Ah, I see. I'll add a TODO to find a better way to do this.

+
+        // All the previous phases have finished building. If anyone's still
+        // waiting for outputs, cut off the wait; we won't be generating them,
+        // at least until a source asset changes.
+        for (var completer in _pendingOutputRequests.values) {
+          completer.complete(null);
+        }
+        _pendingOutputRequests.clear();
+      });
+    }
+  }
 
   /// 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 22 matching lines @@
     _onLogPool.add(input.onLog);
     input.onDone.listen((_) {
       if (!isDirty) _onDoneController.add(null);
     });
 
     for (var group in _groups.values) {
       group.addInput(node);
     }
   }
 
+  // TODO(nweiz): If the input is available when this is called, it's
+  // theoretically possible for it to become unavailable between the call and
+  // the return. If it does so, it won't trigger the rebuilding process. To
+  // avoid this, we should have this and the methods it calls take explicit
+  // callbacks, as in [AssetNode.whenAvailable].
   /// Gets the asset node for an input [id].
   ///
-  /// If an input with that ID cannot be found, returns null.
+  /// If [id] is for a generated or transformed asset, this will wait until it
+  /// has been created and return it. This means that the returned asset will
+  /// always be [AssetState.AVAILABLE].
+  /// 
+  /// If the input cannot be found, returns null.
   Future<AssetNode> getInput(AssetId id) {
-    return newFuture(() {
+    return syncFuture(() {
       if (id.package != cascade.package) return cascade.graph.getAssetNode(id);
-      if (_inputs.containsKey(id)) return _inputs[id].input;
-      return null;
+      if (_previous != null) return _previous.getOutput(id);
+      if (!_inputs.containsKey(id)) return null;
+
+      var input = _inputs[id].input;
+      return input.whenAvailable((_) => input).catchError((error) {
+        if (error is! AssetNotFoundException || error.id != id) throw error;
+        // Retry in case the input was replaced.
+        return getInput(id);
+      });
     });
   }
 
   /// Gets the asset node for an output [id].
   ///
-  /// If an output with that ID cannot be found, returns null.
+  /// If [id] is for a generated or transformed asset, this will wait until it
+  /// has been created and return it. This means that the returned asset will
+  /// always be [AssetState.AVAILABLE].
+  /// 
+  /// If the output cannot be found, returns null.
   Future<AssetNode> getOutput(AssetId id) {
-    return newFuture(() {
+    return syncFuture(() {
       if (id.package != cascade.package) return cascade.graph.getAssetNode(id);
-      if (!_outputs.containsKey(id)) return null;
-      var output = _outputs[id].output;
-      output.force();
-      return output;
+      if (_outputs.containsKey(id)) {
+        var output = _outputs[id].output;
+        // If the requested output is available, we can just return it.
+        if (output.state.isAvailable) return output;
+
+        // If the requested output exists but isn't yet available, wait to see if

[Bob Nystrom, 2014/03/06 23:59:22]

Long line.

[nweiz, 2014/03/07 00:20:19]

Done.

+        // it becomes available. If it's removed before becoming available, try
+        // again, since it could be generated again.
+        output.force();
+        return output.whenAvailable((_) => output).catchError((error) {
+          if (error is! AssetNotFoundException) throw error;
+          return getOutput(id);
+        });
+      }
+
+      // If neither this phase nor the previous phases are dirty, the requested
+      // output won't be generated and we can safely return null.
+      if (!_isTransitivelyDirty) return null;
+
+      // Otherwise, store a completer for the asset node. If it's generated in
+      // the future, we'll complete this completer.
+      var completer = _pendingOutputRequests.putIfAbsent(id,
+          () => new Completer.sync());
+      return completer.future;
     });
   }
 
   /// Set this phase's transformers to [transformers].
   void updateTransformers(Iterable transformers) {
     var actualTransformers = transformers.where((op) => op is Transformer);
     _transformers.clear();
     _transformers.addAll(actualTransformers);
     for (var input in _inputs.values) {
       input.updateTransformers(actualTransformers);
@@ skipping 34 matching lines @@
     for (var input in _inputs.values) {
       input.forceAllTransforms();
     }
   }
 
   /// Add a new phase after this one.
   ///
   /// This may only be called on a phase with no phase following it.
   Phase addPhase() {
     assert(_next == null);
-    _next = new Phase._(cascade, _location, _index + 1);
+    _next = new Phase._(cascade, _location, _index + 1, this);
     for (var output in _outputs.values.toList()) {
       // Remove [output]'s listeners because now they should get the asset from
       // [_next], rather than this phase. Any transforms consuming [output] will
       // be re-run and will consume the output from the new final phase.
       output.removeListeners();
     }
     return _next;
   }
 
   /// Mark this phase as removed.
   ///
   /// This will remove all the phase's outputs and all following phases.
   void remove() {
+    _previous._next = null;
     removeFollowing();
     for (var input in _inputs.values.toList()) {
       input.remove();
     }
     for (var group in _groups.values) {
       group.remove();
     }
     _onAssetController.close();
     _onLogPool.close();
   }
@@ skipping 33 matching lines @@
   /// Emit [asset] as an output of this phase.
   ///
   /// This should be called after [_handleOutput], so that collisions are
   /// resolved.
   void _emit(AssetNode asset) {
     if (_next != null) {
       _next.addInput(asset);
     } else {
       _onAssetController.add(asset);
     }
+    _providePendingAsset(asset);
+  }
+
+  /// Provide an asset to a pending [getOutput] call.
+  void _providePendingAsset(AssetNode asset) {
+    // If anyone's waiting for this asset, provide it to them.
+    var request = _pendingOutputRequests.remove(asset.id);
+    if (request == null) return;
+
+    if (asset.state.isAvailable) {
+      request.complete(asset);
+      return;
+    }
+
+    // A lazy asset may be emitted while still dirty. If so, we wait until it's
+    // either available or removed before trying again to access it.
+    assert(asset.state.isDirty);
+    asset.force();
+    asset.whenStateChanges().then((state) {
+      if (state.isRemoved) return getOutput(asset.id);
+      return asset;
+    }).then(request.complete).catchError(request.completeError);
   }
 
   String toString() => "phase $_location.$_index";
 }