Move Barback to a more thoroughly push-based model.

pkg/barback/lib/src/phase_forwarder.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_forwarder;
 
 import 'dart:async';
 
 import 'asset_node.dart';
 
@@ skipping 35 matching lines @@
   ///
   /// This will be null if the asset is not being forwarded.
   AssetNode get output => _outputController.node;
   AssetNodeController _outputController;
 
   /// A stream that emits an event whenever [this] starts producing a final
   /// forwarded asset.
   ///
   /// Whenever this stream emits an event, the value will be identical to
   /// [output].
-  Stream<AssetNode> get onForwarding => _onForwardingController.stream;
-  final _onForwardingController =
-      new StreamController<AssetNode>.broadcast(sync: true);
+  Stream<AssetNode> get onAsset => _onAssetController.stream;
+  final _onAssetController = new StreamController<AssetNode>(sync: true);
 
   PhaseForwarder(this._numChannels);
 
   /// Adds an intermediate forwarded asset to [this].
   ///
   /// [asset] must have the same origin as all other intermediate forwarded
   /// assets.
   void addIntermediateAsset(AssetNode asset) {
     if (_intermediateAssets.isNotEmpty) {
       assert(asset.origin == _intermediateAssets.first.origin);
@@ skipping 10 matching lines @@
   }
 
   /// Mark this forwarder as removed.
   ///
   /// This will remove [output] if it exists.
   void remove() {
     if (_outputController != null) {
       _outputController.setRemoved();
       _outputController = null;
     }
-    _onForwardingController.close();
+    _onAssetController.close();
   }
 
   /// Adjusts [output] to ensure that it accurately reflects the current state
   /// of the intermediate forwarded assets.
   void _adjustOutput() {
     assert(_intermediateAssets.length <= _numChannels);
     assert(!_intermediateAssets.any((asset) => asset.state.isRemoved));
 
     // If there are any channels that haven't forwarded an intermediate asset,
     // we shouldn't forward a final asset. If we are currently, remove
     // it.
     if (_intermediateAssets.length < _numChannels) {
       if (_outputController == null) return;
       _outputController.setRemoved();
       _outputController = null;
       return;
     }
 
     // If there isn't a final asset being forwarded yet, we should forward one.
     // It should be dirty iff any of the intermediate assets are dirty.
     if (_outputController == null) {
       var finalAsset = _intermediateAssets.firstWhere(
           (asset) => asset.state.isDirty,
           orElse: () => _intermediateAssets.first);
       _outputController = new AssetNodeController.from(finalAsset);
-      _onForwardingController.add(output);
+      _onAssetController.add(output);
       return;
     }
 
     // If we're already forwarding a final asset, set it dirty iff any of the
     // intermediate assets are dirty.
     if (_intermediateAssets.any((asset) => asset.state.isDirty)) {
       if (!_outputController.node.state.isDirty) _outputController.setDirty();
     } else {
       if (!_outputController.node.state.isAvailable) {
         _outputController.setAvailable(_intermediateAssets.first.asset);
       }
     }
   }
 }