Fix a barback bug where a closed stream could get an event.

pkg/barback/lib/src/graph/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.graph.phase;
 
 import 'dart:async';
 
 import '../asset/asset_id.dart';
 import '../asset/asset_node.dart';
@@ skipping 95 matching lines @@
 
   /// The previous phase in the cascade, or null if this is the first phase.
   final Phase previous;
 
   /// The subscription to [previous]'s [onStatusChange] stream.
   StreamSubscription _previousStatusSubscription;
 
   /// The subscription to [previous]'s [onAsset] stream.
   StreamSubscription<AssetNode> _previousOnAssetSubscription;
 
+  final _inputSubscriptions = new Set<StreamSubscription>();
+
   /// A map of asset ids to completers for [getInput] requests.
   ///
   /// If an asset node is requested before it's available, we put a completer in
   /// this map to wait for the asset to be generated. If it's not generated, the
   /// completer should complete to `null`.
   final _pendingOutputRequests = new Map<AssetId, Completer<AssetNode>>();
 
   /// Returns all currently-available output assets for this phase.
   Set<AssetNode> get availableOutputs {
     return _outputs.values
@@ skipping 44 matching lines @@
     var forwarder = new PhaseForwarder(
         node, _classifiers.length, _groups.length);
     _forwarders[node.id] = forwarder;
     forwarder.onAsset.listen(_handleOutputWithoutForwarder);
     if (forwarder.output != null) {
       _handleOutputWithoutForwarder(forwarder.output);
     }
 
     _inputOrigins.add(node.origin);
     _inputs.add(node);
-    node.onStateChange.listen((state) {
+    _inputSubscriptions.add(node.onStateChange.listen((state) {
       if (state.isRemoved) {
         _inputOrigins.remove(node.origin);
         _forwarders.remove(node.id).remove();
       }
       _streams.changeStatus(status);
-    });
+    }));
 
     for (var classifier in _classifiers.values) {
       classifier.addInput(node);
     }
   }
 
   // TODO(nweiz): If the output is available when this is called, it's
   // theoretically possible for it to become unavailable between the call and
   // the return. If it does so, it won't trigger the rebuilding process. To
   // avoid this, we should have this and the methods it calls take explicit
@@ skipping 120 matching lines @@
   ///
   /// This will remove all the phase's outputs.
   void remove() {
     for (var classifier in _classifiers.values.toList()) {
       classifier.remove();
     }
     for (var group in _groups.values) {
       group.remove();
     }
     _streams.close();
+    for (var subscription in _inputSubscriptions) {
+      subscription.cancel();
+    }
     if (_previousStatusSubscription != null) {
       _previousStatusSubscription.cancel();
     }
     if (_previousOnAssetSubscription != null) {
       _previousOnAssetSubscription.cancel();
     }
   }
 
   /// Add [asset] as an output of this phase.
   void _handleOutput(AssetNode asset) {
@@ skipping 45 matching lines @@
     assert(asset.state.isDirty);
     asset.force();
     asset.whenStateChanges().then((state) {
       if (state.isRemoved) return getOutput(asset.id);
       return asset;
     }).then(request.complete).catchError(request.completeError);
   }
 
   String toString() => "phase $_location.$_index";
 }