Implement addStream for HttpClientRequest/HttpResponse and propegate all write-errors from the sock…

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>> implements StreamSink<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
@@ skipping 336 matching lines @@
   void add(List<int> data) {
     _writeHeaders();
     if (_ignoreBody || data.length == 0) return;
     if (_chunked) {
       _ChunkedTransformer._addChunk(data, super.add);
     } else {
       super.add(data);
     }
   }
 
+  Future addStream(Stream<List<int>> stream) {
+    _writeHeaders();
+    if (_ignoreBody) return new Future.immediate(this);
+    if (_chunked) {
+      // Transform when chunked.
+      stream = stream.transform(new _ChunkedTransformer(writeEnd: false));
+    }
+    return super.addStream(stream).then((_) => this);
+  }
+
   void close() {
     if (!_headersWritten && !_ignoreBody && headers.chunkedTransferEncoding) {
       // If no body was written, _ignoreBody is false (it's not a HEAD
       // request) and the content-length is unspecified, set contentLength to 0.
       headers.contentLength = 0;
     }
     _writeHeaders();
     if (!_ignoreBody) {
       if (_chunked) {
         _ChunkedTransformer._addChunk([], super.add);
@@ skipping 296 matching lines @@
     headers._finalize();
 
     // Write headers.
     headers._write(this);
     writeCRLF();
   }
 }
 
 
 // Transformer that transforms data to HTTP Chunked Encoding.
-class _ChunkedTransformer implements StreamTransformer<List<int>, List<int>> {
-  final StreamController<List<int>> _controller
-      = new StreamController<List<int>>();
+class _ChunkedTransformer extends StreamEventTransformer<List<int>, List<int>> {
+  final bool writeEnd;
+  _ChunkedTransformer({this.writeEnd: true});
 
-  Stream<List<int>> bind(Stream<List<int>> stream) {
-    var subscription = stream.listen(
-        (data) {
-          if (data.length == 0) return;  // Avoid close on 0-bytes payload.
-          _addChunk(data, _controller.add);
-       },
-       onDone: () {
-          _addChunk([], _controller.add);
-          _controller.close();
-       });
-    return _controller.stream;
+  void handleData(List<int> data, StreamSink<List<int>> sink) {
+    _addChunk(data, sink.add);
+  }
+
+  void handleDone(StreamSink<List<int>> sink) {
+    if (writeEnd) {
+      _addChunk([], sink.add);
+    }
+    sink.close();
   }
 
   static void _addChunk(List<int> data, void add(List<int> data)) {
     add(_chunkHeader(data.length));
     if (data.length > 0) add(data);
     add(_chunkFooter);
   }
 
   static List<int> _chunkHeader(int length) {
     const hexDigits = const [0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
@@ skipping 203 matching lines @@
             }, test: (error) => error is StateError)
             .catchError((error) {
               // We are done with the socket.
               destroy();
               request._onError(error);
             });
 
         return _socket.addStream(stream)
             .catchError((e) {
               destroy();
-              if (e.error is HttpException) throw e;
-              // TODO(ajohnsen): Where to send Socket errors?
+              throw e;
             });
     });
     return request;
   }
 
   Future<Socket> detachSocket() {
     return _streamFuture
         .then((_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()),
               onError: (_) {});
   }
@@ skipping 284 matching lines @@
                     // request is now processed.
                     _subscription.resume();
                   } else {
                     // Close socket, keep-alive not used or body sent before
                     // received data was handled.
                     destroy();
                   }
                 })
                 .catchError((e) {
                   destroy();
-                  if (e.error is HttpException) throw e;
-                  // TODO(ajohnsen): Where to send Socket errors?
+                  throw e;
                 });
           });
           response._ignoreBody = request.method == "HEAD";
           response._httpRequest = request;
           _httpServer._handleRequest(request);
         },
         onDone: () {
           destroy();
         },
         onError: (error) {
@@ skipping 372 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;
 }