Move common streams in barback to their own class.

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';
 import 'asset_node.dart';
 import 'errors.dart';
 import 'group_runner.dart';
 import 'log.dart';
 import 'multiset.dart';
+import 'node_streams.dart';
 import 'phase_forwarder.dart';
 import 'phase_input.dart';
 import 'phase_output.dart';
-import 'stream_pool.dart';
 import 'transformer.dart';
 import 'transformer_group.dart';
 import 'utils.dart';
 
 /// One phase in the ordered series of transformations in an [AssetCascade].
 ///
 /// Each phase can access outputs from previous phases and can in turn pass
 /// outputs to later phases. Phases are processed strictly serially. All
 /// transforms in a phase will be complete before moving on to the next phase.
 /// Within a single phase, all transforms will be run in parallel.
@@ skipping 41 matching lines @@
   /// that input isn't consumed by any transformers, it will be forwarded
   /// through the PhaseInput. However, it's possible that it was consumed by a
   /// group, and so shouldn't be forwarded through the phase as a whole.
   ///
   /// In order to detect whether an output has been forwarded through a group or
   /// a PhaseInput, we must be able to distinguish it from other outputs with
   /// the same id. To do so, we check if its origin is in [_inputOrigins]. If
   /// so, it's been forwarded unmodified.
   final _inputOrigins = new Multiset<AssetNode>();
 
-  /// 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);
-
-  /// 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>.broadcast(sync: true);
+  /// The streams exposed by this phase.
+  final _streams = new NodeStreams();
+  Stream get onDone => _streams.onDone;
+  Stream<AssetNode> get onAsset => _streams.onAsset;
+  Stream<LogEntry> get onLog => _streams.onLog;
 
   /// 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 subscription to [previous]'s [onAsset] stream.
   StreamSubscription<AssetNode> _previousOnAssetSubscription;
 
   /// A map of asset ids to completers for [getInput] requests.
@@ skipping 14 matching lines @@
   // 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);
+        if (!isDirty) _streams.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) {
         completer.complete(null);
       }
@@ skipping 24 matching lines @@
       _handleOutputWithoutForwarder(forwarder.output);
     }
 
     _inputOrigins.add(node.origin);
     var input = new PhaseInput(this, node, "$_location.$_index");
     _inputs[node.id] = input;
     input.input.whenRemoved(() {
       _inputOrigins.remove(node.origin);
       _inputs.remove(node.id);
       _forwarders.remove(node.id).remove();
-      if (!isDirty) _onDoneController.add(null);
+      if (!isDirty) _streams.onDoneController.add(null);
     });
     input.onAsset.listen(_handleOutput);
-    _onLogPool.add(input.onLog);
+    _streams.onLogPool.add(input.onLog);
     input.onDone.listen((_) {
-      if (!isDirty) _onDoneController.add(null);
+      if (!isDirty) _streams.onDoneController.add(null);
     });
 
     input.updateTransformers(_transformers);
 
     for (var group in _groups.values) {
       group.addInput(node);
     }
   }
 
   // TODO(nweiz): If the output is available when this is called, it's
@@ skipping 53 matching lines @@
         .toSet();
     var oldGroups = _groups.keys.toSet();
     for (var removed in oldGroups.difference(newGroups)) {
       _groups.remove(removed).remove();
     }
 
     for (var added in newGroups.difference(oldGroups)) {
       var runner = new GroupRunner(cascade, added, "$_location.$_index");
       _groups[added] = runner;
       runner.onAsset.listen(_handleOutput);
-      _onLogPool.add(runner.onLog);
+      _streams.onLogPool.add(runner.onLog);
       runner.onDone.listen((_) {
-        if (!isDirty) _onDoneController.add(null);
+        if (!isDirty) _streams.onDoneController.add(null);
       });
       for (var input in _inputs.values) {
         runner.addInput(input.input);
       }
     }
 
     for (var forwarder in _forwarders.values) {
       forwarder.updateTransformers(_transformers.length, _groups.length);
     }
   }
@@ skipping 27 matching lines @@
   /// Mark this phase as removed.
   ///
   /// This will remove all the phase's outputs.
   void remove() {
     for (var input in _inputs.values.toList()) {
       input.remove();
     }
     for (var group in _groups.values) {
       group.remove();
     }
-    _onAssetController.close();
-    _onLogPool.close();
+    _streams.close();
     if (_previousOnDoneSubscription != null) {
       _previousOnDoneSubscription.cancel();
     }
     if (_previousOnAssetSubscription != null) {
       _previousOnAssetSubscription.cancel();
     }
   }
 
   /// Add [asset] as an output of this phase.
   void _handleOutput(AssetNode asset) {
@@ skipping 18 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) {
-    _onAssetController.add(asset);
+    _streams.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";
 }