Revert commits r33138, r33135, and r33134.

pkg/barback/lib/src/group_runner.dart

Index: pkg/barback/lib/src/group_runner.dart
diff --git a/pkg/barback/lib/src/group_runner.dart b/pkg/barback/lib/src/group_runner.dart
index 217d54b7f8fb6a9038d21548a538791b2c85bf41..2e8941621ef48bce473c470e7ff3c7a0a3656494 100644
--- a/pkg/barback/lib/src/group_runner.dart
+++ b/pkg/barback/lib/src/group_runner.dart
@@ -8,7 +8,6 @@ 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';
@@ -51,7 +50,7 @@ class GroupRunner {
   ///
   /// [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);
@@ -99,6 +98,9 @@ class GroupRunner {
     // 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);