Move common streams in barback to their own class.

pkg/barback/lib/src/transform_node.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.transform_node;
 
 import 'dart:async';
 
 import 'asset.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'declaring_transform.dart';
 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';
 
 /// Describes a transform on a set of assets and its relationship to the build
 /// dependency graph.
 ///
 /// Keeps track of whether it's dirty and needs to be run and which assets it
 /// depends on.
 class TransformNode {
@@ skipping 45 matching lines @@
 
   /// The controller that's used to pass [primary] through [this] if it's not
   /// consumed or overwritten.
   ///
   /// This needs an intervening controller to ensure that the output can be
   /// marked dirty when determining whether [this] will consume or overwrite it,
   /// and be marked removed if it does. [_passThroughController] will be null
   /// 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.
   ///
   /// Defaults to `false`. This is not meaningful unless [_state] is
   /// [_State.APPLIED] or [_State.DECLARED].
   bool _consumePrimary = false;
 
   /// The set of output ids that [transformer] declared it would emit.
   ///
   /// This is only non-null if [transformer] is a [DeclaringTransformer] and its
   /// [declareOutputs] has been run successfully.
   Set<AssetId> _declaredOutputs;
 
   TransformNode(this.phase, Transformer transformer, AssetNode primary,
       this._location)
       : transformer = transformer,
         primary = primary,
         deferred = transformer is LazyTransformer ||
             (transformer is DeclaringTransformer && primary.deferred) {
     _forced = !deferred;
 
-    _onLogPool.add(_onLogController.stream);
-
     _primarySubscription = primary.onStateChange.listen((state) {
       if (state.isRemoved) {
         remove();
       } else {
         if (state.isDirty && !deferred) primary.force();
         // If this is deferred but applying, that means it must have been
         // forced, so we should ensure its input remains forced as well.
         if (deferred && _forced && _state == _State.APPLYING) primary.force();
         _dirty();
       }
@@ skipping 14 matching lines @@
   /// node.
   TransformInfo get info => new TransformInfo(transformer, primary.id);
 
   /// Marks this transform as removed.
   ///
   /// This causes all of the transform's outputs to be marked as removed as
   /// well. Normally this will be automatically done internally based on events
   /// 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();
     _clearOutputs();
     if (_passThroughController != null) {
       _passThroughController.setRemoved();
       _passThroughController = null;
     }
   }
 
@@ skipping 69 matching lines @@
     }).then((isPrimary) {
       if (_isRemoved) return null;
       if (isPrimary) {
         if (!deferred) primary.force();
         return _declareOutputs().then((_) {
           if (_isRemoved) return;
           if (_forced) {
             _apply();
           } else {
             _state = _State.DECLARED;
-            _onDoneController.add(null);
+            _streams.onDoneController.add(null);
           }
         });
       }
 
       _emitPassThrough();
       _state = _State.NOT_PRIMARY;
-      _onDoneController.add(null);
+      _streams.onDoneController.add(null);
     });
   }
 
   /// Runs [transform.declareOutputs] and emits the resulting assets as dirty
   /// assets.
   Future _declareOutputs() {
     if (transformer is! DeclaringTransformer) return new Future.value();
 
     var controller = new DeclaringTransformController(this);
     return syncFuture(() {
@@ skipping 13 matching lines @@
         phase.cascade.reportError(new InvalidOutputException(info, id));
       }
 
       if (!_declaredOutputs.contains(primary.id)) _emitPassThrough();
 
       for (var id in _declaredOutputs) {
         var controller = _forced
             ? 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;
       phase.cascade.reportError(_wrapException(error, stackTrace));
     });
   }
 
   /// Applies this transform.
   void _apply() {
     assert(!_isRemoved);
@@ skipping 23 matching lines @@
         // consume the pass-through asset, we can safely emit it.
         if (_declaredOutputs != null && !_consumePrimary &&
             !_declaredOutputs.contains(primary.id)) {
           _emitPassThrough();
         } else {
           _dontEmitPassThrough();
         }
       }
 
       _state = _State.APPLIED;
-      _onDoneController.add(null);
+      _streams.onDoneController.add(null);
     });
   }
 
   /// Gets the asset for an input [id].
   ///
   /// If an input with [id] cannot be found, throws an [AssetNotFoundException].
   Future<Asset> getInput(AssetId id) {
     return phase.previous.getOutput(id).then((node) {
       // Throw if the input isn't found. This ensures the transformer's apply
       // is exited. We'll then catch this and report it through the proper
       // results stream.
       if (node == null) {
         _missingInputs.add(id);
         throw new AssetNotFoundException(id);
       }
 
       _inputSubscriptions.putIfAbsent(node.id, () {
         return node.onStateChange.listen((state) => _dirty());
       });
 
       return node.asset;
     });
   }
 
   /// Run [Transformer.apply] as soon as [primary] is available.
   ///
   /// 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;
       _state = _State.APPLYING;
       return syncFuture(() => transformer.apply(transformController.transform));
     }).then((_) {
       if (deferred && !_forced && !primary.state.isAvailable) {
         _state = _State.DECLARED;
-        _onDoneController.add(null);
+        _streams.onDoneController.add(null);
         return false;
       }
 
       if (_isRemoved) return false;
       if (_state == _State.NEEDS_APPLY) return false;
       if (_state == _State.DECLARING) return false;
       if (transformController.loggedError) return true;
       _handleApplyResults(transformController);
       return false;
     }).catchError((error, stackTrace) {
@@ skipping 42 matching lines @@
     }
 
     // Store any new outputs or new contents for existing outputs.
     for (var asset in newOutputs) {
       var controller = _outputControllers[asset.id];
       if (controller != null) {
         controller.setAvailable(asset);
       } else {
         var controller = new AssetNodeController.available(asset, this);
         _outputControllers[asset.id] = controller;
-        _onAssetController.add(controller.node);
+        _streams.onAssetController.add(controller.node);
       }
     }
   }
 
   /// Cancels all subscriptions to secondary input nodes.
   void _clearInputSubscriptions() {
     _missingInputs.clear();
     for (var subscription in _inputSubscriptions.values) {
       subscription.cancel();
     }
     _inputSubscriptions.clear();
   }
 
   /// Removes all output assets.
   void _clearOutputs() {
     // Remove all the previously-emitted assets.
     for (var controller in _outputControllers.values) {
       controller.setRemoved();
     }
     _outputControllers.clear();
   }
 
   /// Emit the pass-through asset if it's not being emitted already.
   void _emitPassThrough() {
     assert(!_outputControllers.containsKey(primary.id));
 
     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) {
       _passThroughController.setAvailable(primary.asset);
     }
   }
 
   /// Stop emitting the pass-through asset if it's being emitted already.
   void _dontEmitPassThrough() {
     if (_passThroughController == null) return;
     _passThroughController.setRemoved();
     _passThroughController = null;
   }
 
   BarbackException _wrapException(error, StackTrace stackTrace) {
     if (error is! AssetNotFoundException) {
       return new TransformerException(info, error, stackTrace);
     } else {
       return new MissingInputException(info, error.id);
     }
   }
 
   /// 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));
   }
 
   String toString() =>
     "transform node in $_location for $transformer on $primary";
 }
 
 /// The enum of states that [TransformNode] can be in.
 class _State {
   /// The transform is running [Transformer.isPrimary] followed by
@@ skipping 44 matching lines @@
   ///
   /// This will never transition to another state.
   static final NOT_PRIMARY = const _State._("not primary");
 
   final String name;
 
   const _State._(this.name);
 
   String toString() => name;
 }