Clean up usage of streams in http-parser.

sdk/lib/io/http_parser.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;
 
 // Global constants.
 class _Const {
   // Bytes for "HTTP".
   static const HTTP = const [72, 84, 84, 80];
@@ skipping 216 matching lines @@
  * object will be [:true:] indicating that from now on the protocol is
  * not HTTP anymore and no more callbacks will happen, that is
  * [:dataReceived:] and [:dataEnd:] are not called in this case as
  * there is no more HTTP data. After the upgrade the method
  * [:readUnparsedData:] can be used to read any remaining bytes in the
  * HTTP parser which are part of the protocol the connection is
  * upgrading to. These bytes cannot be processed by the HTTP parser
  * and should be handled according to whatever protocol is being
  * upgraded to.
  */
-class _HttpParser
-    extends Stream<_HttpIncoming>
-    implements StreamConsumer<List<int>> {
+class _HttpParser extends Stream<_HttpIncoming> {
   // State.
   bool _parserCalled = false;
 
   // The data that is currently being parsed.
   Uint8List _buffer;
   int _index;
 
   final bool _requestParser;
   int _state;
   int _httpVersionIndex;
@@ skipping 58 matching lines @@
   StreamSubscription<_HttpIncoming> listen(void onData(_HttpIncoming event),
                                            {Function onError,
                                             void onDone(),
                                             bool cancelOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
-  Future<_HttpParser> addStream(Stream<List<int>> stream) {
+  void listenToStream(Stream<List<int>> stream) {
     // Listen to the stream and handle data accordingly. When a
     // _HttpIncoming is created, _dataPause, _dataResume, _dataDone is
     // given to provide a way of controlling the parser.
     // TODO(ajohnsen): Remove _dataPause, _dataResume and _dataDone and clean up
     // how the _HttpIncoming signals the parser.
-    var completer = new Completer();
     _socketSubscription = stream.listen(
         _onData,
         onError: _onError,
-        onDone: () {
-          completer.complete(this);
-        });
-    return completer.future;
-  }
-
-  Future<_HttpParser> close() {
-    _onDone();
-    return new Future.value(this);
+        onDone: _onDone);
   }
 
   void _parse() {
     try {
       _doParse();
     } catch (e, s) {
       _state = _State.FAILURE;
       _reportError(e, s);
     }
   }
@@ skipping 693 matching lines @@
     }
   }
 
   void _reportError(error, [stackTrace]) {
     if (_socketSubscription != null) _socketSubscription.cancel();
     _state = _State.FAILURE;
     _controller.addError(error, stackTrace);
     _controller.close();
   }
 }