Make StreamSubscription be the active part of a stream.

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 528 matching lines @@
 class _HttpOutboundConsumer implements StreamConsumer {
   final _HttpOutboundMessage _outbound;
   StreamController _controller;
   StreamSubscription _subscription;
   Completer _closeCompleter = new Completer();
   Completer _completer;
 
   _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);
               },
               onError: (error) {
                 if (!_done(error)) {
                   _closeCompleter.completeError(error);
                 }
               });
   }
 
   bool _done([error]) {
     if (_completer == null) return false;
     if (error != null) {
       _completer.completeError(error);
     } else {
       _completer.complete(_outbound);
     }
     _completer = null;
     return true;
   }
 
   Future addStream(var stream) {
     _ensureController();
     _completer = new Completer();
+    Future result = _completer.future;

[Anders Johnsen, 2013/05/22 13:07:08]

Nooo... :)

[floitsch, 2013/05/22 16:26:29]

Apparently this was necessary.
Add comment. (it's not obvious)

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

It should not be necessary any more after I prevented sending events during the
onListen callback (stupid bug). Will change it back.

     _subscription = stream.listen(
         (data) {
           _controller.add(data);
         },
         onDone: () {
           _done();
         },
         onError: (error) {
           _done(error);
         },
         cancelOnError: true);
-    return _completer.future;
+    return result;
   }
 
   Future close() {
     Future closeOutbound() {
       return _outbound._close().then((_) => _outbound);
     }
     if (_controller == null) return closeOutbound();
     _controller.close();
     return _closeCompleter.future.then((_) => closeOutbound());
   }
@@ skipping 1695 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;
 }