First pass at build dependency graph for barback.

pkg/barback/lib/src/asset_graph.dart

Index: pkg/barback/lib/src/asset_graph.dart
diff --git a/pkg/barback/lib/src/asset_graph.dart b/pkg/barback/lib/src/asset_graph.dart
new file mode 100644
index 0000000000000000000000000000000000000000..6c12889cefcea4a11401344a733cdec42b8f6d3e
--- /dev/null
+++ b/pkg/barback/lib/src/asset_graph.dart
@@ -0,0 +1,171 @@
+// 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_graph;
+
+import 'dart:async';
+import 'dart:collection';
+
+import '../barback.dart';
+import '../transformer.dart';

[nweiz, 2013/06/18 23:14:46]

"src" files shouldn't be importing "lib" files. They should be importing the
libraries they need directly.

[Bob Nystrom, 2013/06/20 00:23:59]

Done.

+import 'change_batch.dart';
+import 'phase.dart';
+
+/// The main build dependency manager. For any given input file, it can tell
+/// which output files are affected by it, and vice versa.
+class AssetGraph {
+  final AssetProvider _provider;
+
+  final _phases = <Phase>[];
+
+  Stream<ProcessResult> get results => _resultsController.stream;
+  final _resultsController = new StreamController<ProcessResult>.broadcast();
+
+  /// This holds a future that completes when the build process is complete if
+  /// if it is currently running. Otherwise, it is `null`.
+  Future _processDone;
+
+  ChangeBatch _sourceChanges;
+
+  /// Creates a new [AssetGraph].
+  ///
+  /// It loads source assets using [provider] and then uses [transformerPhases]
+  /// to generate output files from them.
+  //TODO(rnystrom): Better way of specifying transformers and their ordering.
+  AssetGraph(this._provider,
+      Iterable<Iterable<Transformer>> transformerPhases) {
+    // Add phases for each transformer stage.
+    for (var transformers in transformerPhases) {
+      var phase = new Phase(this, _phases.length, transformers.toList());
+      _phases.add(phase);
+    }
+
+    // Each phase writes its outputs as inputs to the next phase after it.
+    // Add a phase at the end for the final outputs of the last phase.
+    _phases.add(new Phase(this, _phases.length, []));
+
+    // Chain them together.
+    for (var i = 0; i < _phases.length - 1; i++){
+      _phases[i].next = _phases[i + 1];
+    }
+  }
+
+  /// 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. If the asset cannot be found, throws
+  /// [AssetNotFoundException].
+  Future<Asset> getAssetById(AssetId id) {
+    return _waitForProcess().then((_) {
+      // Find the latest phase that output this asset.
+      for (var i = _phases.length - 1; i >= 0; i--) {
+        var node = _phases[i].inputs[id];
+        if (node != null) {
+          // By the time we get here, the asset should have been built.
+          assert(node.asset != null);
+          return node.asset;
+        }
+      }
+
+      // Couldn't find it.
+      var error = new AssetNotFoundException(id);
+      reportError(error);
+      throw error;

[nweiz, 2013/06/18 23:14:46]

My thought here was that you could *just* throw when you need to report an
error, then have a [catchError] on [getAssetById] that reports the error to
[_resultsController].

[Bob Nystrom, 2013/06/20 00:23:59]

In most cases, though, we don't want to just unwind to the top of the public API
and bail on an error. I intend barback to be resilient and catch those
internally where it can do something more appropriate.

+    });
+  }
+
+  /// 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) {
+    if (_sourceChanges == null) _sourceChanges = new ChangeBatch();
+    _sourceChanges.update(sources);
+
+    _waitForProcess();
+  }
+
+  /// Removes [removed] from the graph's known set of source assets.
+  void removeSources(Iterable<AssetId> removed) {
+    if (_sourceChanges == null) _sourceChanges = new ChangeBatch();
+    _sourceChanges.remove(removed);
+
+    _waitForProcess();
+  }
+
+  /// Reports a process result with the given error then throws it.
+  void reportError(error) {
+    _resultsController.add(new ProcessResult(error));
+  }
+
+  /// 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;
+    return _processDone = _process().whenComplete(() {

[nweiz, 2013/06/18 23:14:46]

If you're confident this won't emit errors, why are you using [whenComplete]?

[Bob Nystrom, 2013/06/20 00:23:59]

Why not?

[nweiz, 2013/06/20 23:06:08]

Because functionally it's identical to [then], but it communicates to a reader
that you expect that an error might occur here.

[Bob Nystrom, 2013/06/21 00:13:20]

Added a catchError() too here like you suggested.

+      _processDone = null;
+    });
+  }
+
+  /// Starts the background processing. Returns a future that completes when
+  /// all assets have been processed.
+  Future _process() {
+    return _processSourceChanges().then((_) {
+      // 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;
+      }
+
+      // If all phases are done, so are we.
+      if (future == null) return;
+
+      // Process that phase and then loop onto the next.
+      return future.then((_) => _process());
+    });
+  }
+
+  /// Processes the current batch of changes to source assets.
+  Future _processSourceChanges() {
+    // Always pump the event loop. This ensures a bunch of synchronous source
+    // changes are processed in a single batch even when the first one starts
+    // the build process.
+    return new Future(() {
+      if (_sourceChanges == null) return null;
+
+      // Take the current batch to ensure it doesn't get added to while we're
+      // processing it.
+      var changes = _sourceChanges;
+      _sourceChanges = null;
+
+      var updated = new Map<AssetId, Asset>();
+      var futures = [];
+      for (var id in changes.updated) {
+        futures.add(_provider.getAsset(id).then((asset) {
+          updated[id] = asset;
+        }));
+      }
+
+      return Future.wait(futures).then((_) {
+        _phases.first.updateInputs(updated, changes.removed);
+      });
+    });
+  }
+}
+
+/// The build process runs asynchronously in the background. It reports back to
+/// the user be emitting a [Stream] of these objects. Currently, it only emits
+/// errors.
+class ProcessResult {
+  /// The error that occurred.
+  final error;
+
+  ProcessResult(this.error);
+}