Better handling of asset collisions in barback.

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 'dart:collection';
 
 import 'asset.dart';
 import 'asset_cascade.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'errors.dart';
 import 'stream_pool.dart';
 import 'transform_node.dart';
 import 'transformer.dart';
@@ skipping 43 matching lines @@
   /// Whenever an asset is added or modified, we need to asynchronously
   /// determine which transformers can use it as their primary input. We can't
   /// start processing until we know which transformers to run, and this allows
   /// us to wait until we do.
   var _adjustTransformersFutures = new Map<AssetId, Future>();
 
   /// New asset nodes that were added while [_adjustTransformers] was still
   /// being run on an old version of that asset.
   var _pendingNewInputs = new Map<AssetId, AssetNode>();
 
-  /// The ids of assets that are emitted by transforms in this phase.
+  /// A map of output ids to the asset node outputs for those ids.
   ///
-  /// This is used to detect collisions where multiple transforms emit the same
-  /// output.
-  final _outputs = new Set<AssetId>();
+  /// Usually there's only one node for a given output id. However, it's
+  /// possible for multiple transformers to output an asset with the same id. In

[Bob Nystrom, 2013/08/09 22:44:43]

"transformers" -> "transformers in this phase"

[nweiz, 2013/08/12 21:15:44]

Done.

+  /// that case, the first output emitted is passed forward. We keep track of

[Bob Nystrom, 2013/08/09 22:44:43]

"first" -> "chronologically first"

[nweiz, 2013/08/12 21:15:44]

Done.

+  /// the other nodes so that if that output is removed, we know which asset to
+  /// replace it with.
+  final _outputs = new Map<AssetId, Queue<AssetNode>>();
 
   /// A stream that emits an event whenever this phase becomes dirty and needs
   /// to be run.
   ///
   /// This may emit events when the phase was already dirty or while 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 => _onDirtyPool.stream;
   final _onDirtyPool = new StreamPool.broadcast();
 
@@ skipping 217 matching lines @@
 
   /// Applies all currently wired up and dirty transforms.
   Future _processTransforms() {
     // Convert this to a list so we can safely modify _transforms while
     // iterating over it.
     var dirtyTransforms =
         flatten(_transforms.values.map((transforms) => transforms.toList()))
         .where((transform) => transform.isDirty).toList();
     if (dirtyTransforms.isEmpty) return null;
 
-    return Future.wait(dirtyTransforms.map((transform) => transform.apply()))
-        .then((allNewOutputs) {
-      var newOutputs = allNewOutputs.reduce((set1, set2) => set1.union(set2));
+    var collisions = new Set<AssetId>();
+    return Future.wait(dirtyTransforms.map((transform) {
+      return transform.apply().then((outputs) {
+        for (var output in outputs) {
+          if (_outputs.containsKey(output.id)) {
+            _outputs[output.id].add(output);
+            collisions.add(output.id);
+          } else {
+            _outputs[output.id] = new Queue<AssetNode>.from([output]);
+            _next.addInput(output);
+          }
 
-      var collisions = new Set<AssetId>();
-      for (var newOutput in newOutputs) {
-        if (_outputs.contains(newOutput.id)) {
-          collisions.add(newOutput.id);
-        } else {
-          _next.addInput(newOutput);
-          _outputs.add(newOutput.id);
-          newOutput.whenRemoved.then((_) => _outputs.remove(newOutput.id));
+          _handleOutputRemoval(output);
         }
-      }
-
+      });
+    })).then((_) {
       // Report collisions in a deterministic order.
       collisions = collisions.toList();
       collisions.sort((a, b) => a.toString().compareTo(b.toString()));

[Bob Nystrom, 2013/08/09 22:44:43]

This presumes AssetNode.toString() does something well-defined, which is a bit
sketchy.

Add a TODO that it should implement Comparable explicitly.

[nweiz, 2013/08/12 21:15:44]

I just went ahead and implemented comparable.

       for (var collision in collisions) {
+        // Ensure that there's still a collision. It's possible it was resolved
+        // while another transform was running.
+        if (_outputs[collision].length < 2) return;

[Bob Nystrom, 2013/08/09 22:44:43]

"< 2" -> "<= 1"
"return" -> "continue"

[nweiz, 2013/08/12 21:15:44]

Done.

         cascade.reportError(new AssetCollisionException(collision));
-        // TODO(rnystrom): Define what happens after a collision occurs.
       }
     });
   }
+
+  /// Properly resolve collisions when [output] is removed.
+  void _handleOutputRemoval(AssetNode output) {
+    output.whenRemoved.then((_) {
+      var assets = _outputs[output.id];
+      var wasFirst = assets.first == output;
+      assets.remove(output);
+
+      // If this wasn't the first asset, then a collision was resolved,
+      // possibly partially.
+      if (wasFirst) {
+        // If this is the only asset with this id, we're no longer
+        // producing that output. Otherwise, there was a collision, and we
+        // should return whatever asset was emitted after the one that was
+        // just removed.
+        if (assets.isEmpty) {
+          _outputs.remove(output.id);

[Bob Nystrom, 2013/08/09 22:44:43]

How about handling this case first and then exiting the function? So:

output.whenRemoved.then((_) {
  var assets = _outputs[output.id];
  if (assets.length == 1) {
    _outputs.remove(output.id);
   return;
  }

  // Now handle collision edge cases...
});

This is the most common case, so it would be nice to see it first.

[Bob Nystrom, 2013/08/09 22:44:43]

Also, do we need to tell the next phase the input was removed, or is that
handled elsewhere?

[nweiz, 2013/08/12 21:15:44]

Done.
The next phase has subscribed to the input's [onStateChange] feed and will be
notified that way.

+        } else {
+          // Pump the event queue to give [_next] a chance to handle the removal
+          // of its input before getting a new input.
+          newFuture(() => _next.addInput(assets.first));
+        }
+      }
+
+      // If there's still a collision, report it. This lets the user know
+      // if they've successfully resolved the collision or not.
+      if (assets.length > 1) {
+        // Pump the event queue to ensure that the removal of the input triggers
+        // a new build to which we can attach the error.
+        newFuture(() =>
+            cascade.reportError(new AssetCollisionException(output.id)));
+      }
+    });
+  }
 }