Reapply "Active stream subscriptions".

tests/standalone/io/http_parser_test.dart

Index: tests/standalone/io/http_parser_test.dart
diff --git a/tests/standalone/io/http_parser_test.dart b/tests/standalone/io/http_parser_test.dart
index 756bf36be63c4c17e3cce4820000e7741b0e60cb..e6564bdc60ab7fb5c842b60f2f4038d93eab6981 100644
--- a/tests/standalone/io/http_parser_test.dart
+++ b/tests/standalone/io/http_parser_test.dart
@@ -202,6 +202,24 @@ class HttpParserTest {
       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;
@@ -228,21 +246,9 @@ class HttpParserTest {
             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;