Make barback's onAsset streams broadcast rather than single-subscriber.

pkg/barback/lib/src/phase.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;
 
 import 'dart:async';
 
 import 'asset_cascade.dart';
 import 'asset_id.dart';
@@ skipping 73 matching lines @@
   /// soon as [isDirty] flips from `true` to `false`.
   Stream get onDone => _onDoneController.stream;
   final _onDoneController = new StreamController.broadcast(sync: true);
 
   /// A stream that emits any new assets emitted by [this].
   ///
   /// Assets are emitted synchronously to ensure that any changes are thoroughly
   /// propagated as soon as they occur. Only a phase with no [next] phase will
   /// emit assets.
   Stream<AssetNode> get onAsset => _onAssetController.stream;
-  final _onAssetController = new StreamController<AssetNode>(sync: true);
+  final _onAssetController =
+      new StreamController<AssetNode>.broadcast(sync: true);
 
   /// Whether [this] is dirty and still has more processing to do.
   ///
   /// A phase is considered dirty if any of the previous phases in the same
   /// cascade are dirty, since those phases could emit an asset that this phase
   /// will then need to process.
   bool get isDirty => (_previous != null && _previous.isDirty) ||
       _inputs.values.any((input) => input.isDirty) ||
       _groups.values.any((group) => group.isDirty);
 
   /// A stream that emits an event whenever any transforms in this phase logs
   /// an entry.
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
   /// The previous phase in the cascade, or null if this is the first phase.
   final Phase _previous;
 
   /// The subscription to [_previous]'s [onDone] stream.
   StreamSubscription _previousOnDoneSubscription;
 
-  /// The phase after this one.
-  ///
-  /// Outputs from this phase will be passed to it.
-  Phase get next => _next;
-  Phase _next;
+  /// The subscription to [_previous]'s [onAsset] stream.
+  StreamSubscription<AssetNode> _previousOnAssetSubscription;
 
   /// A map of asset ids to completers for [getInput] 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 _pendingOutputRequests = new Map<AssetId, Completer<AssetNode>>();
 
   /// Returns all currently-available output assets for this phase.
   Set<AssetNode> get availableOutputs {
     return _outputs.values
         .map((output) => output.output)
         .where((node) => node.state.isAvailable)
         .toSet();
   }
 
   // TODO(nweiz): Rather than passing the cascade and the phase everywhere,
   // create an interface that just exposes [getInput]. Emit errors via
   // [AssetNode]s.
   Phase(AssetCascade cascade, String location)
       : this._(cascade, location, 0);
 
   Phase._(this.cascade, this._location, this._index, [this._previous]) {
     if (_previous != null) {
+      _previousOnAssetSubscription = _previous.onAsset.listen(addInput);
       _previousOnDoneSubscription = _previous.onDone.listen((_) {
         if (!isDirty) _onDoneController.add(null);
       });
     }
 
     onDone.listen((_) {
       // All the previous phases have finished building. If anyone's still
       // waiting for outputs, cut off the wait; we won't be generating them,
       // at least until a source asset changes.
       for (var completer in _pendingOutputRequests.values) {
@@ skipping 159 matching lines @@
 
     for (var input in _inputs.values) {
       input.forceAllTransforms();
     }
   }
 
   /// Add a new phase after this one.
   ///
   /// This may only be called on a phase with no phase following it.
   Phase addPhase() {
-    assert(_next == null);
-    _next = new Phase._(cascade, _location, _index + 1, this);
+    var next = new Phase._(cascade, _location, _index + 1, this);
     for (var output in _outputs.values.toList()) {
       // Remove [output]'s listeners because now they should get the asset from
-      // [_next], rather than this phase. Any transforms consuming [output] will
+      // [next], rather than this phase. Any transforms consuming [output] will
       // be re-run and will consume the output from the new final phase.
       output.removeListeners();
     }
-    return _next;
+    return next;
   }
 
   /// Mark this phase as removed.
   ///
-  /// This will remove all the phase's outputs and all following phases.
+  /// This will remove all the phase's outputs.
   void remove() {
-    if (_previous != null) _previous._next = null;
-    removeFollowing();
     for (var input in _inputs.values.toList()) {
       input.remove();
     }
     for (var group in _groups.values) {
       group.remove();
     }
     _onAssetController.close();
     _onLogPool.close();
     if (_previousOnDoneSubscription != null) {
       _previousOnDoneSubscription.cancel();
     }
-  }
-
-  /// Remove all phases after this one.
-  void removeFollowing() {
-    if (_next == null) return;
-    _next.remove();
-    _next = null;
+    if (_previousOnAssetSubscription != null) {
+      _previousOnAssetSubscription.cancel();
+    }
   }
 
   /// Add [asset] as an output of this phase.
   void _handleOutput(AssetNode asset) {
     if (_inputOrigins.contains(asset.origin)) {
       _forwarders[asset.id].addIntermediateAsset(asset);
     } else {
       _handleOutputWithoutForwarder(asset);
     }
   }
@@ skipping 12 matching lines @@
 
     var exception = _outputs[asset.id].collisionException;
     if (exception != null) cascade.reportError(exception);
   }
 
   /// Emit [asset] as an output of this phase.
   ///
   /// This should be called after [_handleOutput], so that collisions are
   /// resolved.
   void _emit(AssetNode asset) {
-    if (_next != null) {
-      _next.addInput(asset);
-    } else {
-      _onAssetController.add(asset);
-    }
+    _onAssetController.add(asset);
     _providePendingAsset(asset);
   }
 
   /// Provide an asset to a pending [getOutput] call.
   void _providePendingAsset(AssetNode asset) {
     // If anyone's waiting for this asset, provide it to them.
     var request = _pendingOutputRequests.remove(asset.id);
     if (request == null) return;
 
     if (asset.state.isAvailable) {
       request.complete(asset);
       return;
     }
 
     // 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.
     assert(asset.state.isDirty);
     asset.force();
     asset.whenStateChanges().then((state) {
       if (state.isRemoved) return getOutput(asset.id);
       return asset;
     }).then(request.complete).catchError(request.completeError);
   }
 
   String toString() => "phase $_location.$_index";
 }