Refactor the Transform classes.

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 'package:source_maps/span.dart';
-
 import 'asset.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'declaring_transform.dart';
 import 'errors.dart';
 import 'lazy_transformer.dart';
 import 'log.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 33 matching lines @@
   /// processing transforms. Events are emitted synchronously to ensure that the
   /// dirty state is thoroughly propagated as soon as any assets are changed.
   Stream get onDirty => _onDirtyController.stream;
   final _onDirtyController = new StreamController.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 => _onLogController.stream;
-  final _onLogController = new StreamController<LogEntry>.broadcast(sync: true);
+  Stream<LogEntry> get onLog => _onLogPool.stream;
+  final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
   TransformNode(this.phase, Transformer transformer, this.primary,
       this._location)
       : transformer = transformer,
         _isLazy = transformer is LazyTransformer {
     _primarySubscription = primary.onStateChange.listen((state) {
       if (state.isRemoved) {
         remove();
       } else {
         _dirty();
@@ skipping 108 matching lines @@
         // If the node was removed before it could be loaded, treat it as though
         // it never existed and throw a MissingInputException.
         throw new MissingInputException(info, id);
       });
     });
   }
 
   /// Applies the transform so that it produces concrete (as opposed to lazy)
   /// outputs.
   Future<Set<AssetNode>> _applyImmediate() {
-    var newOutputs = new AssetSet();
-    var transform = new Transform(this, newOutputs, _log);
+    var transformController = new TransformController(this);
+    _onLogPool.add(transformController.onLog);
 
-    return syncFuture(() => transformer.apply(transform)).then((_) {
+    return syncFuture(() {
+      return transformer.apply(transformController.transform);
+    }).whenComplete(transformController.close).then((_) {
       if (_isDirty) return new Set();
 
+      var newOutputs = new AssetSet.from(transformController.outputs);

[Bob Nystrom, 2014/02/27 23:41:18]

outputs already is an AssetSet. Is there a reason to copy it here?

[nweiz, 2014/02/28 01:19:08]

It just felt kind of weird to modify another class's instance variable. I'm
happy to change it to access the set directly if yuo think that's cleaner.

[Bob Nystrom, 2014/02/28 18:56:39]

I look at it more like claiming ownership. It feels a bit weird to me too, but
in this case the owning object is discarded immediately after this. Not copying
is a little faster and (more important to me) a little less code.

       // Any ids that are for a different package are invalid.
       var invalidIds = newOutputs
           .map((asset) => asset.id)
           .where((id) => id.package != phase.cascade.package)
           .toSet();
       for (var id in invalidIds) {
         newOutputs.removeId(id);
         // TODO(nweiz): report this as a warning rather than a failing error.
         phase.cascade.reportError(new InvalidOutputException(info, id));
       }
@@ skipping 17 matching lines @@
         }
       }
 
       return brandNewOutputs;
     });
   }
 
   /// Applies the transform in declarative mode so that it produces lazy
   /// outputs.
   Future<Set<AssetNode>> _declareLazy() {
-    var newIds = new Set();
-    var transform = new DeclaringTransform(this, newIds, _log);
+    var transformController = new DeclaringTransformController(this);
 
     return syncFuture(() {
-      return (transformer as LazyTransformer).declareOutputs(transform);
-    }).then((_) {
+      return (transformer as LazyTransformer)
+          .declareOutputs(transformController.transform);
+    }).whenComplete(transformController.close).then((_) {
       if (_isDirty) return new Set();
 
+      var newIds = new Set.from(transformController.outputIds);

[Bob Nystrom, 2014/02/27 23:41:18]

Ditto. Why copy the set here?

       var invalidIds =
           newIds.where((id) => id.package != phase.cascade.package).toSet();
       for (var id in invalidIds) {
         newIds.remove(id);
         // TODO(nweiz): report this as a warning rather than a failing error.
         phase.cascade.reportError(new InvalidOutputException(info, id));
       }
 
       // Remove outputs that used to exist but don't anymore.
       for (var id in _outputControllers.keys.toList()) {
@@ skipping 10 matching lines @@
           var controller = new AssetNodeController.lazy(id, force, this);
           _outputControllers[id] = controller;
           brandNewOutputs.add(controller.node);
         }
       }
 
       return brandNewOutputs;
     });
   }
 
-  void _log(AssetId asset, LogLevel level, String message, Span span) {
-    // If the log isn't already associated with an asset, use the primary.
-    if (asset == null) asset = primary.id;
-    var entry = new LogEntry(info, asset, level, message, span);
-    _onLogController.add(entry);
-  }
-
   String toString() =>
     "transform node in $_location for $transformer on $primary";
 }