Reapply "Active stream subscriptions".

sdk/lib/io/http_impl.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.
 
 part of dart.io;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
   Stream<List<int>> _stream;
@@ skipping 544 matching lines @@
   StreamController _controller;
   StreamSubscription _subscription;
   Completer _closeCompleter = new Completer();
   Completer _completer;
   bool _socketError = false;
 
   _HttpOutboundConsumer(_HttpOutboundMessage this._outbound);
 
   void _cancel() {
     if (_subscription != null) {
-      _subscription.cancel();
+      StreamSubscription subscription = _subscription;
+      _subscription = null;
+      subscription.cancel();
     }
   }
 
   _ensureController() {
     if (_controller != null) return;
     _controller = new StreamController(onPause: () => _subscription.pause(),
                                        onResume: () => _subscription.resume(),
                                        onListen: () => _subscription.resume(),
                                        onCancel: _cancel);
     _outbound._addStream(_controller.stream)
@@ skipping 1760 matching lines @@
 
 
 class _RedirectInfo implements RedirectInfo {
   const _RedirectInfo(int this.statusCode,
                       String this.method,
                       Uri this.location);
   final int statusCode;
   final String method;
   final Uri location;
 }