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

pkg/barback/lib/src/group_runner.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.group_runner;
 
 import 'dart:async';
 
 import 'asset_cascade.dart';
 import 'asset_node.dart';
@@ skipping 19 matching lines @@
   bool get isDirty {
     // Just check the last phase, since it will check all the previous phases
     // itself.
     return _phases.last.isDirty;
   }
 
   /// A stream that emits an event whenever [this] is no longer dirty.
   ///
   /// This is synchronous in order to guarantee that it will emit an event as
   /// soon as [isDirty] flips from `true` to `false`.
-  Stream get onDone => _onDoneController.stream;
-  final _onDoneController = new StreamController.broadcast(sync: true);
+  Stream get onDone => _onDone;
+  Stream _onDone;
 
   /// 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.
-  Stream<AssetNode> get onAsset => _onAssetPool.stream;
-  final _onAssetPool = new StreamPool<AssetNode>();
+  Stream<AssetNode> get onAsset => _onAsset;
+  Stream<AssetNode> _onAsset;
 
   /// A stream that emits an event whenever any transforms in this group logs
   /// an entry.
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
   GroupRunner(AssetCascade cascade, this._group, this._location) {
     _addPhase(new Phase(cascade, _location), _group.phases.first);
     for (var phase in _group.phases.skip(1)) {
       _addPhase(_phases.last.addPhase(), phase);
     }
+
+    _onAsset = _phases.last.onAsset;
+    _onDone = _phases.last.onDone;
   }
 
   /// Add a phase with [contents] to [this]'s list of phases.
   ///
   /// [contents] should be an inner [Iterable] from a [TransformGroup.phases]
   /// value.
   void _addPhase(Phase phase, Iterable contents) {
     _phases.add(phase);
-    _onAssetPool.add(phase.onAsset);
     _onLogPool.add(phase.onLog);
-    phase.onDone.listen((_) {
-      if (!isDirty) _onDoneController.add(null);
-    });
     phase.updateTransformers(contents);
   }
 
   /// Force all [LazyTransformer]s' transforms in this group to begin producing
   /// concrete assets.
   void forceAllTransforms() {
     for (var phase in _phases) {
       phase.forceAllTransforms();
     }
   }
 
   /// Adds a new asset as an input for this group.
   void addInput(AssetNode node) {
     _phases.first.addInput(node);
   }
 
   /// Removes this group and all sub-phases within it.
   void remove() {
-    _phases.first.remove();
+    for (var phase in _phases) {
+      phase.remove();
+    }
   }
 
   String toString() => "group in phase $_location for $_group";
 }