Revert commits r33138, r33135, and r33134.

pkg/barback/lib/src/phase_forwarder.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_forwarder;
 
 import 'dart:async';
 
 import 'asset_node.dart';
-import 'asset_node_set.dart';
 
 /// A class that takes care of forwarding assets within a phase.
 ///
 /// Each phase contains one or more channels that process its input assets. The
 /// non-grouped transformers for that phase are one such channel, and each
 /// [TransformerGroup] in that phase is another. For each input asset, each
 /// channel individually decides whether to forward that asset based on whether
 /// that channel uses it. If a channel does decide to forward an asset, we call
 /// that forwarded asset an "intermediate forwarded asset" to distinguish it
 /// from the output of a [PhaseForwarder].
@@ skipping 13 matching lines @@
   ///
   /// Each group is a channel, along with one channel for the [PhaseInput] that
   /// handles all the transformers.
   set numChannels(int value) {
     _numChannels = value;
     _adjustOutput();
   }
   int _numChannels;
 
   /// The intermediate forwarded assets.
-  final _intermediateAssets = new AssetNodeSet();
+  final _intermediateAssets = new Set<AssetNode>();
 
   /// The final forwarded asset.
   ///
   /// This will be null if the asset is not being forwarded.
   AssetNode get output => _outputController.node;
   AssetNodeController _outputController;
 
   /// A stream that emits an event whenever [this] starts producing a final
   /// forwarded asset.
   ///
   /// Whenever this stream emits an event, the value will be identical to
   /// [output].
   Stream<AssetNode> get onForwarding => _onForwardingController.stream;
   final _onForwardingController =
       new StreamController<AssetNode>.broadcast(sync: true);
 
   PhaseForwarder(this._numChannels);
 
   /// Adds an intermediate forwarded asset to [this].
   ///
   /// [asset] must have the same origin as all other intermediate forwarded
   /// assets.
   void addIntermediateAsset(AssetNode asset) {
     if (_intermediateAssets.isNotEmpty) {
       assert(asset.origin == _intermediateAssets.first.origin);
     }
 
     _intermediateAssets.add(asset);
-    asset.onStateChange.listen((_) => _adjustOutput());
+
+    asset.onStateChange.listen((state) {
+      if (state.isRemoved) _intermediateAssets.remove(asset);
+      _adjustOutput();
+    });
 
     _adjustOutput();
   }
 
   /// Mark this forwarder as removed.
   ///
   /// This will remove [output] if it exists.
   void remove() {
     if (_outputController != null) {
       _outputController.setRemoved();
@@ skipping 33 matching lines @@
     // intermediate assets are dirty.
     if (_intermediateAssets.any((asset) => asset.state.isDirty)) {
       if (!_outputController.node.state.isDirty) _outputController.setDirty();
     } else {
       if (!_outputController.node.state.isAvailable) {
         _outputController.setAvailable(_intermediateAssets.first.asset);
       }
     }
   }
 }