Move Barback to a more thoroughly push-based model.

pkg/barback/lib/src/asset_cascade.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.asset_cascade;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'asset.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'log.dart';
-import 'build_result.dart';
 import 'cancelable_future.dart';
 import 'errors.dart';
 import 'package_graph.dart';
 import 'phase.dart';
 import 'stream_pool.dart';
 import 'transformer.dart';
-import 'utils.dart';
 
 /// The asset cascade for an individual package.
 ///
 /// This keeps track of which [Transformer]s are applied to which assets, and
 /// re-runs those transformers when their dependencies change. The transformed
 /// asset nodes are accessible via [getAssetNode].
 ///
 /// A cascade consists of one or more [Phases], each of which has one or more
 /// [Transformer]s that run in parallel, potentially on the same inputs. The
 /// inputs of the first phase are the source assets for this cascade's package.
@@ skipping 13 matching lines @@
 
   /// Futures for source assets that are currently being loaded.
   ///
   /// These futures are cancelable so that if an asset is updated after a load
   /// has been kicked off, the previous load can be ignored in favor of a new
   /// one.
   final _loadingSources = new Map<AssetId, CancelableFuture<Asset>>();
 
   final _phases = <Phase>[];
 
-  /// A stream that emits a [BuildResult] each time the build is completed,
-  /// whether or not it succeeded.
-  ///
-  /// If an unexpected error in barback itself occurs, it will be emitted
-  /// through this stream's error channel.
-  Stream<BuildResult> get results => _resultsController.stream;
-  final _resultsController = new StreamController<BuildResult>.broadcast();
-
   /// A stream that emits any errors from the cascade or the transformers.
   ///
   /// This emits errors as they're detected. If an error occurs in one part of
   /// the cascade, unrelated parts will continue building.
-  ///
-  /// This will not emit programming errors from barback itself. Those will be
-  /// emitted through the [results] stream's error channel.
   Stream<BarbackException> get errors => _errorsController.stream;
-  final _errorsController = new StreamController<BarbackException>.broadcast();
-
-  /// A stream that emits an event whenever this cascade becomes dirty.
-  ///
-  /// After this stream emits an event, [results] will emit an event once the
-  /// cascade is no longer dirty.
-  ///
-  /// This may emit events when the cascade was already dirty. 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].
-  final _onDirtyController = new StreamController.broadcast(sync: true);
+  final _errorsController =
+      new StreamController<BarbackException>.broadcast(sync: true);
 
   /// A stream that emits an event whenever any transforms in this cascade logs
   /// an entry.
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
-  /// The errors that have occurred since the current build started.
+  /// Whether [this] is dirty and still has more processing to do.
+  bool get isDirty => _phases.any((phase) => phase.isDirty);
+
+  /// A stream that emits an event whenever [this] is no longer dirty.
   ///
-  /// This will be empty if no build is occurring.
-  Queue<BarbackException> _accumulatedErrors;
-
-  /// The number of errors that have been logged since the current build
-  /// started.
-  int _numLogErrors;
-
-  /// A future that completes when the currently running build process finishes.
-  ///
-  /// If no build it in progress, is `null`.
-  Future _processDone;
-
-  /// Whether any source assets have been updated or removed since processing
-  /// last began.
-  var _newChanges = false;
+  /// 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<bool>.broadcast(sync: true);

[Bob Nystrom, 2014/03/05 22:13:25]

Ditch the <bool>.

[nweiz, 2014/03/06 00:29:08]

Done.

 
   /// Returns all currently-available output assets from this cascade.
   AssetSet get availableOutputs =>
     new AssetSet.from(_phases.last.availableOutputs.map((node) => node.asset));
 
+  /// A map of asset ids to completers for [getAssetNode] 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 _pendingAssetRequests = new Map<AssetId, Queue<Completer<AssetNode>>>();
+
   /// Creates a new [AssetCascade].
   ///
   /// It loads source assets within [package] using [provider].
   AssetCascade(this.graph, this.package) {
-    _onDirtyPool.add(_onDirtyController.stream);
-    _addPhase(new Phase(this, [], package));
-
-    // Keep track of logged errors so we can know that the build failed.
-    onLog.listen((entry) {
-      if (entry.level == LogLevel.ERROR) {
-        // TODO(nweiz): keep track of stack chain.
-        _accumulatedErrors.add(
-            new TransformerException(entry.transform, entry.message, null));
-      }
-    });
+    _addPhase(new Phase(this, package));
   }
 
   /// Gets the asset identified by [id].
   ///
   /// 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 asset cannot be found, returns null.
   Future<AssetNode> getAssetNode(AssetId id) {
     assert(id.package == package);
 
     // TODO(rnystrom): Waiting for the entire build to complete is unnecessary
     // in some cases. Should optimize:
     // * [id] may be generated before the compilation is finished. We should
     //   be able to quickly check whether there are any more in-place
     //   transformations that can be run on it. If not, we can return it early.
     // * If [id] has never been generated and all active transformers provide
     //   metadata about the file names of assets it can emit, we can prove that
     //   none of them can emit [id] and fail early.
     return _phases.last.getOutput(id).then((node) {
-      // If the requested asset is available, we can just return it.
-      if (node != null && node.state.isAvailable) return node;
+      if (node != null) {
+        // If the requested asset is available, we can just return it.
+        if (node.state.isAvailable) return node;
 
-      if (_processDone != null) {
-        // If there's a build running, that build might generate the asset, so
-        // we wait for it to complete and then try again.
-        return _processDone.then((_) => getAssetNode(id));
+        // If the requested asset exists but isn't yet available, wait to see if
+        // it becomes available. If it's removed before becoming available, try
+        // again, since it could be generated again.
+        node.force();
+        return node.whenAvailable((_) => node).catchError((error) {
+          if (error is! AssetNotFoundException) throw error;
+          return getAssetNode(id);
+        });
       }
 
-      // If the asset hasn't been built and nothing is building now, the asset
-      // won't be generated, so we return null.
-      return null;
+      // If the cascade isn't dirty, the phase won't generate the requested
+      // asset in the future.
+      if (!isDirty) return null;
+
+      // If the cascade is dirty, store a completer for the asset node. If it's
+      // generated in the future, we'll complete this completer.
+      var completers = _pendingAssetRequests.putIfAbsent(id, () => new Queue());

[Bob Nystrom, 2014/03/05 22:13:25]

Why do you need a queue here? Can you just reuse the previous completer for this
asset?

[nweiz, 2014/03/06 00:29:08]

Done.

+      completers.add(new Completer.sync());
+      return completers.last.future;
     });
   }
 
   /// Adds [sources] to the graph's known set of source assets.
   ///
   /// Begins applying any transforms that can consume any of the sources. If a
   /// given source is already known, it is considered modified and all
   /// transforms that use it will be re-applied.
   void updateSources(Iterable<AssetId> sources) {
     for (var id in sources) {
@@ skipping 42 matching lines @@
   /// [Transformer]s or [TransformerGroup]s.
   void updateTransformers(Iterable<Iterable> transformersIterable) {
     var transformers = transformersIterable.toList();
 
     for (var i = 0; i < transformers.length; i++) {
       if (_phases.length > i) {
         _phases[i].updateTransformers(transformers[i]);
         continue;
       }
 
-      _addPhase(_phases.last.addPhase(transformers[i]));
+      var phase = _phases.last.addPhase();
+      _addPhase(phase);
+      phase.updateTransformers(transformers[i]);
     }
 
     if (transformers.length == 0) {
       _phases.last.updateTransformers([]);
     } else if (transformers.length < _phases.length) {
       _phases[transformers.length - 1].removeFollowing();
       _phases.removeRange(transformers.length, _phases.length);
     }
   }
 
   /// Force all [LazyTransformer]s' transforms in this cascade to begin
   /// producing concrete assets.
   void forceAllTransforms() {
     for (var phase in _phases) {
       phase.forceAllTransforms();
     }
   }
 
   void reportError(BarbackException error) {
-    _accumulatedErrors.add(error);
     _errorsController.add(error);
   }
 
-  /// Add [phase] to the end of [_phases] and watch its [onDirty] stream.
+  /// Add [phase] to the end of [_phases] and watch its streams.
   void _addPhase(Phase phase) {
-    _onDirtyPool.add(phase.onDirty);
     _onLogPool.add(phase.onLog);
-    phase.onDirty.listen((_) {
-      _newChanges = true;
-      _waitForProcess();
+    phase.onAsset.listen(_providePendingAsset);
+
+    phase.onDone.listen((_) {
+      if (isDirty) return;
+
+      // This cascade has finished building. If anyone's still waiting for
+      // assets, cut off the wait; we won't be generating them, at least until a
+      // source asset changes.
+      _pendingAssetRequests.forEach((_, completers) {
+        for (var completer in completers) {
+          completer.complete(null);
+        }
+      });
+      _pendingAssetRequests.clear();
+      _onDoneController.add(null);
     });
+
     _phases.add(phase);
   }
 
-  /// Starts the build process asynchronously if there is work to be done.
-  ///
-  /// Returns a future that completes with the background processing is done.
-  /// If there is no work to do, returns a future that completes immediately.
-  /// All errors that occur during processing will be caught (and routed to the
-  /// [results] stream) before they get to the returned future, so it is safe
-  /// to discard it.
-  Future _waitForProcess() {
-    if (_processDone != null) return _processDone;
+  /// Provide an asset to a pending [getAssetNode] call.
+  void _providePendingAsset(AssetNode asset) {
+    // If anyone's waiting for this asset, provide it to them.
+    var requests = _pendingAssetRequests.remove(asset.id);
+    if (requests == null) return;
 
-    _accumulatedErrors = new Queue();
-    _numLogErrors = 0;
-    return _processDone = _process().then((_) {
-      // Report the build completion.
-      // TODO(rnystrom): Put some useful data in here.
-      _resultsController.add(
-          new BuildResult(_accumulatedErrors));
-      _processDone = null;
-      _accumulatedErrors = null;
-    });
-  }
-
-  /// Starts the background processing.
-  ///
-  /// Returns a future that completes when all assets have been processed.
-  Future _process() {
-    _newChanges = false;
-    return newFuture(() {
-      // Find the first phase that has work to do and do it.
-      var future;
-      for (var phase in _phases) {
-        future = phase.process();
-        if (future != null) break;
+    for (var request in requests) {
+      if (asset.state.isAvailable) {

[Bob Nystrom, 2014/03/05 22:13:25]

This seems weird to me. Doesn't it make more sense to hoist this out of the
loop? It's the same asset for every request. Then you'd:

1. If the asset is available, complete all of the requests.
2. Otherwise, wait for the state to change and retry.

[nweiz, 2014/03/06 00:29:08]

N/A since I got rid of the queue.

+        request.complete(asset);
+        continue;
       }
 
-      // If all phases are done and no new updates have come in, we're done.
-      if (future == null) {
-        // If changes have come in, start over.
-        if (_newChanges) return _process();
-
-        // Otherwise, everything is done.
-        return null;
-      }
-
-      // Process that phase and then loop onto the next.
-      return future.then((_) => _process());
-    });
+      // 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. We
+      // retry the entire [getAsset] process because the state of the graph may
+      // have changed dramatically by the time it's available.
+      assert(asset.state.isDirty);
+      asset.force();
+      asset.whenStateChanges()
+          .then((_) => getAssetNode(asset.id))
+          .then(request.complete)
+          .catchError(request.completeError);
+    }
   }
 
   String toString() => "cascade for $package";
 }