Revert commits r33138, r33135, and r33134.

pkg/barback/lib/src/group_runner.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.group_runner;
 
 import 'dart:async';
 
 import 'asset_cascade.dart';
 import 'asset_node.dart';
-import 'asset_node_set.dart';
 import 'log.dart';
 import 'phase.dart';
 import 'stream_pool.dart';
 import 'transformer_group.dart';
 
 /// A class that processes all of the phases in a single transformer group.
 ///
 /// A group takes many inputs, processes them, and emits many outputs.
 class GroupRunner {
   /// The group this runner runs.
@@ skipping 22 matching lines @@
   Stream<LogEntry> get onLog => _onLogPool.stream;
   final _onLogPool = new StreamPool<LogEntry>.broadcast();
 
   // TODO(nweiz): move to a more push-based way of propagating outputs and get
   // rid of this. Once that's done, see if we can unify GroupRunner and
   // AssetCascade.
   /// The set of outputs that has been returned by [process].
   ///
   /// [process] is expected to only return new outputs, so this is used to
   /// ensure that it does so.
-  final _alreadyEmittedOutputs = new AssetNodeSet();
+  final _alreadyEmittedOutputs = new Set<AssetNode>();
 
   GroupRunner(AssetCascade cascade, this._group, this._location) {
     var lastPhase = new Phase(cascade, _group.phases.first, _location);
     _phases.add(lastPhase);
     for (var phase in _group.phases.skip(1)) {
       lastPhase = lastPhase.addPhase(phase);
       _phases.add(lastPhase);
     }
 
     for (var phase in _phases) {
@@ skipping 27 matching lines @@
   Future<Set<AssetNode>> process() {
     // Process the first phase that needs to do work.
     for (var phase in _phases) {
       var future = phase.process();
       if (future != null) return future.then((_) => process());
     }
 
     // If we get here, all phases are done processing.
     var newOutputs = _phases.last.availableOutputs
         .difference(_alreadyEmittedOutputs);
+    for (var output in newOutputs) {
+      output.whenRemoved(() => _alreadyEmittedOutputs.remove(output));
+    }
     _alreadyEmittedOutputs.addAll(newOutputs);
 
     return new Future.value(newOutputs);
   }
 
   String toString() => "group in phase $_location for $_group";
 }