Make StreamSubscription be the active part of a stream.

sdk/lib/io/http_impl.dart

Index: sdk/lib/io/http_impl.dart
diff --git a/sdk/lib/io/http_impl.dart b/sdk/lib/io/http_impl.dart
index c749cebd26919281acf0b576c50d0dbdfd58f954..7b512e6ea5a9c98cf352baf3b14f8bdcd017ea13 100644
--- a/sdk/lib/io/http_impl.dart
+++ b/sdk/lib/io/http_impl.dart
@@ -546,28 +546,28 @@ class _HttpOutboundConsumer implements StreamConsumer {
   _HttpOutboundConsumer(_HttpOutboundMessage this._outbound);
 
   void _onPause() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
+    _subscription.pause();
+  }
+  void _onResume() {
+    _subscription.resume();
   }
 
-  void _onListen() {
-    if (!_controller.hasListener && _subscription != null) {
-      _subscription.cancel();
+  void _onCancel() {
+    if (_subscription != null) {
+      StreamSubscription subscription = _subscription;
+      _subscription = null;
+      subscription.cancel();
     }
   }
 
   _ensureController() {
     if (_controller != null) return;
     _controller = new StreamController(onPause: _onPause,
-                                       onResume: _onPause,
-                                       onListen: _onListen,
-                                       onCancel: _onListen);
+                                       onResume: _onResume,
+                                       onCancel: _onCancel);
     _outbound._addStream(_controller.stream)
         .then((_) {
-                _onListen();  // Make sure we unsubscribe.
+                _onCancel();  // Make sure we unsubscribe.
                 _done();
                 _closeCompleter.complete(_outbound);
               },