Merge IO v2 branch to bleeding edge

pkg/http/lib/src/io_client.dart

Index: pkg/http/lib/src/io_client.dart
diff --git a/pkg/http/lib/src/io_client.dart b/pkg/http/lib/src/io_client.dart
index 47fb1f04dc946eb616166189345104a0e35196b8..9bbb7ee00a4f5113ed9d25fb0097261b1358e13f 100644
--- a/pkg/http/lib/src/io_client.dart
+++ b/pkg/http/lib/src/io_client.dart
@@ -24,67 +24,38 @@ class IOClient extends BaseClient {
   Future<StreamedResponse> send(BaseRequest request) {
     var stream = request.finalize();
 
-    var completer = new Completer<StreamedResponse>();
-    var connection = _inner.openUrl(request.method, request.url);
-    bool completed = false;
-    connection.followRedirects = request.followRedirects;
-    connection.maxRedirects = request.maxRedirects;
-    connection.onError = (e) {
-      async.then((_) {
-        // TODO(nweiz): issue 4974 means that any errors that appear in the
-        // onRequest or onResponse callbacks get passed to onError. If the
-        // completer has already fired, we want to re-throw those exceptions
-        // to the top level so that they aren't silently ignored.
-        if (completed) throw e;
-
-        completed = true;
-        completer.completeError(e);
-      });
-    };
-
-    var pipeCompleter = new Completer();
-    connection.onRequest = (underlyingRequest) {
-      underlyingRequest.contentLength = request.contentLength;
-      underlyingRequest.persistentConnection = request.persistentConnection;
+    return _inner.openUrl(request.method, request.url).then((ioRequest) {
+      ioRequest.followRedirects = request.followRedirects;
+      ioRequest.maxRedirects = request.maxRedirects;
+      ioRequest.contentLength = request.contentLength;
+      ioRequest.persistentConnection = request.persistentConnection;
       request.headers.forEach((name, value) {
-        underlyingRequest.headers.set(name, value);
+        ioRequest.headers.set(name, value);
       });
-
-      chainToCompleter(
-          stream.pipe(wrapOutputStream(underlyingRequest.outputStream)),
-          pipeCompleter);
-    };
-
-    connection.onResponse = (response) {
+      return Future.wait([stream.pipe(ioRequest), ioRequest.response])
+          .then((list) => list[1]);
+    }).then((response) {
       var headers = {};
       response.headers.forEach((key, values) {
         headers[key] = values.join(',');
       });
 
-      if (completed) return;
-
-      completed = true;
-      completer.complete(new StreamedResponse(
-          wrapInputStream(response.inputStream),
+      return new StreamedResponse(
+          response,
           response.statusCode,
           response.contentLength,
           request: request,
           headers: headers,
           isRedirect: response.isRedirect,
           persistentConnection: response.persistentConnection,
-          reasonPhrase: response.reasonPhrase));
-    };
-
-    return Future.wait([
-      completer.future,
-      pipeCompleter.future
-    ]).then((values) => values.first);
+          reasonPhrase: response.reasonPhrase);
+    });
   }
 
   /// Closes the client. This terminates all active connections. If a client
   /// remains unclosed, the Dart process may not terminate.
   void close() {
-    if (_inner != null) _inner.shutdown(force: true);
+    if (_inner != null) _inner.close(force: true);
     _inner = null;
   }
 }