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

sdk/lib/io/io_stream_consumer.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;
 
 /**
  * Helper class to wrap a [StreamConsumer<List<int>, T>] and provide utility
  * functions for writing to the StreamConsumer directly. The [IOSink]
  * buffers the input given by [add] and [addString] and will delay a [consume]
@@ skipping 67 matching lines @@
   Future<T> get done {
     _controller;
     return _pipeFuture.then((_) => this);
   }
 
   StreamController<List<int>> get _controller {
     if (_controllerInstance == null) {
       _controllerInstance = new StreamController<List<int>>(
           onPauseStateChange: _onPauseStateChange,
           onSubscriptionStateChange: _onSubscriptionStateChange);
-      _pipeFuture = _controller.stream.pipe(_target);
+      _pipeFuture = _controller.stream.pipe(_target).then((_) => this);
     }
     return _controllerInstance;
   }
 
   bool get _isBound => _bindSubscription != null;
 
   void _onPauseStateChange() {
     _paused = _controller.isPaused;
     if (_controller.isPaused) {
       _pause();
@@ skipping 33 matching lines @@
       }
     }
   }
 
   Future<T> _fillFromStream(Stream<List<int>> stream, {unbind: false}) {
     _controller;
     Completer<T> unbindCompleter;
     if (unbind) {
       unbindCompleter = new Completer<T>();
     }
+    completeUnbind([error]) {
+      if (unbindCompleter == null) return;
+      var tmp = unbindCompleter;
+      unbindCompleter = null;
+      if (error == null) {
+        _bindSubscription = null;
+        tmp.complete();
+      } else {
+        tmp.completeError(error);
+      }
+    }
     _bindSubscription = stream.listen(
         _controller.add,
         onDone: () {
-          _bindSubscription = null;
           if (unbind) {
-            if (unbindCompleter != null) {
-              unbindCompleter.complete(null);
-              unbindCompleter = null;
-            }
+            completeUnbind();
           } else {
             _controller.close();
           }
         },
         onError: _controller.signalError);
     if (_paused) _pause();
     if (unbind) {
-      _pipeFuture.catchError((error) {
-        if (unbindCompleter != null) {
-          unbindCompleter.completeError(error);
-          unbindCompleter = null;
-        }
-      });
+      _pipeFuture
+          .then((_) => completeUnbind(),
+                onError: (error) => completeUnbind(error));
       return unbindCompleter.future;
     } else {
-      return _pipeFuture;
+      return _pipeFuture.then((_) => this);
     }
   }
 }