Move Barback to a more thoroughly push-based model.

pkg/barback/lib/src/phase_output.dart

Index: pkg/barback/lib/src/phase_output.dart
diff --git a/pkg/barback/lib/src/phase_output.dart b/pkg/barback/lib/src/phase_output.dart
index 50d39c3638d5aa0fe413a2e690cd5ceb65b1562e..180715661105cbf87c254d5a8ffc99ed2cf57ce1 100644
--- a/pkg/barback/lib/src/phase_output.dart
+++ b/pkg/barback/lib/src/phase_output.dart
@@ -7,12 +7,10 @@ library barback.phase_output;
 import 'dart:async';
 import 'dart:collection';
 
-import 'asset_cascade.dart';
 import 'asset_forwarder.dart';
 import 'asset_node.dart';
 import 'errors.dart';
 import 'phase.dart';
-import 'utils.dart';
 
 /// A class that handles a single output of a phase.
 ///
@@ -38,7 +36,7 @@ class PhaseOutput {
   /// A stream that emits an [AssetNode] each time this output starts forwarding
   /// a new asset.
   Stream<AssetNode> get onAsset => _onAssetController.stream;
-  final _onAssetController = new StreamController<AssetNode>();
+  final _onAssetController = new StreamController<AssetNode>(sync: true);
 
   /// The assets for this output.
   ///
@@ -105,10 +103,8 @@ class PhaseOutput {
       // If there's still a collision, report it. This lets the user know if
       // they've successfully resolved the collision or not.
       if (_assets.length > 1) {
-        // Pump the event queue to ensure that the removal of the input triggers
-        // a new build to which we can attach the error.
         // TODO(nweiz): report this through the output asset.
-        newFuture(() => _phase.cascade.reportError(collisionException));
+        _phase.cascade.reportError(collisionException);
       }
     });
   }