Transition barback's infrastructure to an aggregate-based model.

pkg/barback/test/package_graph/transform/concurrency_test.dart

Index: pkg/barback/test/package_graph/transform/concurrency_test.dart
diff --git a/pkg/barback/test/package_graph/transform/concurrency_test.dart b/pkg/barback/test/package_graph/transform/concurrency_test.dart
index 3dd2da2f39011f33235cac6a301671db8da519cc..5b0d2e9709beabf82686353bc5f67a0698568187 100644
--- a/pkg/barback/test/package_graph/transform/concurrency_test.dart
+++ b/pkg/barback/test/package_graph/transform/concurrency_test.dart
@@ -505,4 +505,38 @@ main() {
     expectAsset("app|foo.in", "foo");
     buildShouldSucceed();
   });
+
+  test("re-runs if the primary input is invalidated before accessing", () {
+    var transformer1 = new RewriteTransformer("txt", "mid");
+    var transformer2 = new RewriteTransformer("mid", "out");
+
+    initGraph([
+      "app|foo.txt"
+    ], {"app": [
+      [transformer1],
+      [transformer2]
+    ]});
+
+    transformer2.pausePrimaryInput();
+    updateSources(["app|foo.txt"]);
+
+    // Wait long enough to ensure that transformer1 has completed and
+    // transformer2 has started.
+    schedule(pumpEventQueue);
+
+    // Update the source again so that transformer1 invalidates the primary
+    // input of transformer2.
+    transformer1.pauseApply();
+    modifyAsset("app|foo.txt", "new foo");
+    updateSources(["app|foo.txt"]);
+
+    transformer2.resumePrimaryInput();
+    transformer1.resumeApply();
+
+    expectAsset("app|foo.out", "new foo.mid.out");
+    buildShouldSucceed();
+
+    expect(transformer1.numRuns, completion(equals(2)));
+    expect(transformer2.numRuns, completion(equals(2)));
+  });
 }