Clean up usage of streams in http-parser.

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.
 
 library dart.io;
 
 import "package:expect/expect.dart";
 import "dart:async";
 import "dart:collection";
 import "dart:convert";
@@ skipping 52 matching lines @@
       var port2 = new ReceivePort();
 
       String method;
       Uri uri;
       _HttpHeaders headers;
       int contentLength;
       int bytesReceived;
       int unparsedBytesReceived;
       bool upgraded;
 
-      controller.stream.pipe(httpParser);
+      httpParser.listenToStream(controller.stream);
       var subscription = httpParser.listen((incoming) {
         method = incoming.method;
         uri = incoming.uri;
         headers = incoming.headers;
         upgraded = incoming.upgraded;
         Expect.equals(upgrade, upgraded);
 
         if (!chunked) {
           Expect.equals(expectedTransferLength, incoming.transferLength);
         } else {
@@ skipping 67 matching lines @@
 
   static void _testParseInvalidRequest(String request) {
     _HttpParser httpParser;
     bool errorCalled;
     StreamController controller;
 
     void reset() {
       httpParser = new _HttpParser.requestParser();
       controller = new StreamController(sync: true);
       var port = new ReceivePort();
-      controller.stream.pipe(httpParser);
+      httpParser.listenToStream(controller.stream);
       var subscription = httpParser.listen((incoming) {
         Expect.fail("Expected request");
       });
       subscription.onError((e) {
         errorCalled = true;
       });
       subscription.onDone(() {
         port.close();
         Expect.isTrue(errorCalled);
       });
@@ skipping 42 matching lines @@
       bool dataEndCalled;
       bool dataEndClose;
       int statusCode;
       String reasonPhrase;
       _HttpHeaders headers;
       int contentLength;
       int bytesReceived;
       httpParser = new _HttpParser.responseParser();
       controller = new StreamController(sync: true);
       var port = new ReceivePort();
-      controller.stream.pipe(httpParser);
+      httpParser.listenToStream(controller.stream);
       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);
@@ skipping 65 matching lines @@
 
   static void _testParseInvalidResponse(String response, [bool close = false]) {
     void testWrite(List<int> requestData, [int chunkSize = -1]) {
       _HttpParser httpParser = new _HttpParser.responseParser();
       StreamController controller = new StreamController(sync: true);
       bool errorCalled = false;;
 
       if (chunkSize == -1) chunkSize = requestData.length;
 
       var port = new ReceivePort();
-      controller.stream.pipe(httpParser);
+      httpParser.listenToStream(controller.stream);
       var subscription = httpParser.listen((incoming) {
         incoming.listen(
           (data) {},
           onError: (e) {
             Expect.isFalse(errorCalled);
             errorCalled = true;
           });
       });
       subscription.onError((e) {
         Expect.isFalse(errorCalled);
@@ skipping 570 matching lines @@
 0123456789012345678901234567890\r
 0\r\n\r\n""";
     _testParseInvalidResponse(response);
   }
 }
 
 
 void main() {
   HttpParserTest.runAllTests();
 }