Move common streams in barback to their own class.

pkg/barback/lib/src/transform_node.dart

Index: pkg/barback/lib/src/transform_node.dart
diff --git a/pkg/barback/lib/src/transform_node.dart b/pkg/barback/lib/src/transform_node.dart
index e139e1133a3ca25e3229f660115b546e401d2ebc..9a8a370099502c82eacf7f610b1b773ecfa4cc4e 100644
--- a/pkg/barback/lib/src/transform_node.dart
+++ b/pkg/barback/lib/src/transform_node.dart
@@ -14,8 +14,8 @@ import 'declaring_transformer.dart';
 import 'errors.dart';
 import 'lazy_transformer.dart';
 import 'log.dart';
+import 'node_streams.dart';
 import 'phase.dart';
-import 'stream_pool.dart';
 import 'transform.dart';
 import 'transformer.dart';
 import 'utils.dart';
@@ -81,37 +81,17 @@ class TransformNode {
   /// if the asset is not being passed through.
   AssetNodeController _passThroughController;
 
-  /// 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.
-  Stream<AssetNode> get onAsset => _onAssetController.stream;
-  final _onAssetController =
-      new StreamController<AssetNode>.broadcast(sync: true);
-
-  /// A stream that emits an event whenever this transform logs an entry.
-  ///
-  /// This is synchronous because error logs can cause the transform to fail, so
-  /// we need to ensure that their processing isn't delayed until after the
-  /// transform or build has finished.
-  Stream<LogEntry> get onLog => _onLogPool.stream;
-  final _onLogPool = new StreamPool<LogEntry>.broadcast();
-
-  /// A controller for log entries emitted by this node.
-  final _onLogController = new StreamController<LogEntry>.broadcast(sync: true);
+  /// The asset node for this transform.
+  final _streams = new NodeStreams();
+  Stream get onDone => _streams.onDone;
+  Stream<AssetNode> get onAsset => _streams.onAsset;
+  Stream<LogEntry> get onLog => _streams.onLog;
 
   /// The current state of [this].
   var _state = _State.DECLARING;
 
   /// Whether [this] has been marked as removed.
-  bool get _isRemoved => _onAssetController.isClosed;
+  bool get _isRemoved => _streams.onAssetController.isClosed;
 
   /// Whether the most recent run of this transform has declared that it
   /// consumes the primary input.
@@ -134,8 +114,6 @@ class TransformNode {
             (transformer is DeclaringTransformer && primary.deferred) {
     _forced = !deferred;
 
-    _onLogPool.add(_onLogController.stream);
-
     _primarySubscription = primary.onStateChange.listen((state) {
       if (state.isRemoved) {
         remove();
@@ -170,9 +148,7 @@ class TransformNode {
   /// from the primary input, but it's possible for a transform to no longer be
   /// valid even if its primary input still exists.
   void remove() {
-    _onLogController.close();
-    _onAssetController.close();
-    _onDoneController.close();
+    _streams.close();
     _primarySubscription.cancel();
     _phaseSubscription.cancel();
     _clearInputSubscriptions();
@@ -262,14 +238,14 @@ class TransformNode {
             _apply();
           } else {
             _state = _State.DECLARED;
-            _onDoneController.add(null);
+            _streams.onDoneController.add(null);
           }
         });
       }
 
       _emitPassThrough();
       _state = _State.NOT_PRIMARY;
-      _onDoneController.add(null);
+      _streams.onDoneController.add(null);
     });
   }
 
@@ -303,7 +279,7 @@ class TransformNode {
             ? new AssetNodeController(id, this)
             : new AssetNodeController.lazy(id, force, this);
         _outputControllers[id] = controller;
-        _onAssetController.add(controller.node);
+        _streams.onAssetController.add(controller.node);
       }
     }).catchError((error, stackTrace) {
       if (_isRemoved) return;
@@ -347,7 +323,7 @@ class TransformNode {
       }
 
       _state = _State.APPLIED;
-      _onDoneController.add(null);
+      _streams.onDoneController.add(null);
     });
   }
 
@@ -377,7 +353,7 @@ class TransformNode {
   /// Returns whether or not an error occurred while running the transformer.
   Future<bool> _runApply() {
     var transformController = new TransformController(this);
-    _onLogPool.add(transformController.onLog);
+    _streams.onLogPool.add(transformController.onLog);
 
     return primary.whenAvailable((_) {
       if (_isRemoved) return null;
@@ -386,7 +362,7 @@ class TransformNode {
     }).then((_) {
       if (deferred && !_forced && !primary.state.isAvailable) {
         _state = _State.DECLARED;
-        _onDoneController.add(null);
+        _streams.onDoneController.add(null);
         return false;
       }
 
@@ -449,7 +425,7 @@ class TransformNode {
       } else {
         var controller = new AssetNodeController.available(asset, this);
         _outputControllers[asset.id] = controller;
-        _onAssetController.add(controller.node);
+        _streams.onAssetController.add(controller.node);
       }
     }
   }
@@ -479,7 +455,7 @@ class TransformNode {
     if (_consumePrimary) return;
     if (_passThroughController == null) {
       _passThroughController = new AssetNodeController.from(primary);
-      _onAssetController.add(_passThroughController.node);
+      _streams.onAssetController.add(_passThroughController.node);
     } else if (primary.state.isDirty) {
       _passThroughController.setDirty();
     } else if (!_passThroughController.node.state.isAvailable) {
@@ -504,7 +480,7 @@ class TransformNode {
 
   /// Emit a warning about the transformer on [id].
   void _warn(String message) {
-    _onLogController.add(
+    _streams.onLogController.add(
         new LogEntry(info, primary.id, LogLevel.WARNING, message, null));
   }