Make StreamSubscription be the active part of a stream.

tests/standalone/io/http_parser_test.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.
 
 import "package:expect/expect.dart";
 import 'dart:async';
 import 'dart:math';
 import 'dart:typed_data';
 import 'dart:isolate';
 import 'dart:uri';
@@ skipping 186 matching lines @@
       bool dataEndClose;
       int statusCode;
       String reasonPhrase;
       HttpHeaders headers;
       int contentLength;
       int bytesReceived;
       httpParser = new _HttpParser.responseParser();
       controller = new StreamController();
       var port = new ReceivePort();
       controller.stream.pipe(httpParser);
+      int doneCallCount = 0;
+      // Called when done parsing entire message and done parsing body.
+      // Only executed when both are done.
+      void whenDone() {
+        doneCallCount++;
+        if (doneCallCount < 2) return;
+        Expect.equals(expectedVersion, headers.protocolVersion);
+        Expect.equals(expectedStatusCode, statusCode);
+        Expect.equals(expectedReasonPhrase, reasonPhrase);
+        Expect.isTrue(headersCompleteCalled);
+        Expect.equals(expectedBytesReceived, bytesReceived);
+        if (!upgrade) {
+          Expect.isTrue(dataEndCalled);
+          if (close) Expect.isTrue(dataEndClose);
+          Expect.equals(dataEndClose, connectionClose);
+        }
+      };
+
       var subscription = httpParser.listen((incoming) {
         port.close();
         statusCode = incoming.statusCode;
         reasonPhrase = incoming.reasonPhrase;
         headers = incoming.headers;
         Expect.isFalse(headersCompleteCalled);
         if (!chunked && !close) {
           Expect.equals(expectedTransferLength, incoming.transferLength);
         } else {
           Expect.equals(-1, incoming.transferLength);
         }
         if (expectedHeaders != null) {
           expectedHeaders.forEach((String name, String value) {
             Expect.equals(value, headers[name][0]);
           });
         }
         Expect.equals(upgrade, httpParser.upgrade);
         headersCompleteCalled = true;
         incoming.listen(
             (List<int> data) {
               Expect.isTrue(headersCompleteCalled);
               bytesReceived += data.length;
             },
             onDone: () {
               dataEndCalled = true;
               dataEndClose = close;
+              whenDone();
             });
-      });
-
-      subscription.onDone(() {
-        Expect.equals(expectedVersion, headers.protocolVersion);
-        Expect.equals(expectedStatusCode, statusCode);
-        Expect.equals(expectedReasonPhrase, reasonPhrase);
-        Expect.isTrue(headersCompleteCalled);
-        Expect.equals(expectedBytesReceived, bytesReceived);
-        if (!upgrade) {
-          Expect.isTrue(dataEndCalled);
-          if (close) Expect.isTrue(dataEndClose);
-          Expect.equals(dataEndClose, connectionClose);
-        }
-      });
+      }, onDone: whenDone);
 
       headersCompleteCalled = false;
       dataEndCalled = false;
       dataEndClose = null;
       statusCode = -1;
       reasonPhrase = null;
       headers = null;
       bytesReceived = 0;
     }
 
@@ skipping 609 matching lines @@
 0123456789012345678901234567890\r
 0\r\n\r\n""";
     _testParseInvalidResponse(response);
   }
 }
 
 
 void main() {
   HttpParserTest.runAllTests();
 }