Update pkg/http to use the new async APIs.

pkg/http/lib/src/base_request.dart

 // Copyright (c) 2012, 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 base_request;
 
 import 'dart:async';
 import 'dart:io';
 import 'dart:isolate';
 import 'dart:uri';
 
+import 'byte_stream.dart';
 import 'client.dart';
 import 'streamed_response.dart';
+import 'utils.dart';
 
 /// The base class for HTTP requests.
 ///
 /// Subclasses of [BaseRequest] can be constructed manually and passed to
 /// [BaseClient.send], which allows the user to provide fine-grained control
 /// over the request properties. However, usually it's easier to use convenience
 /// methods like [get] or [BaseClient.get].
 abstract class BaseRequest {
   /// The HTTP method of the request. Most commonly "GET" or "POST", less
   /// commonly "HEAD", "PUT", or "DELETE". Non-standard method names are also
@@ skipping 52 matching lines @@
 
   /// Whether the request has been finalized.
   bool get finalized => _finalized;
   bool _finalized = false;
 
   /// Creates a new HTTP request.
   BaseRequest(this.method, this.url)
     : headers = <String>{};
 
   /// Finalizes the HTTP request in preparation for it being sent. This freezes
-  /// all mutable fields and returns an [InputStream] that should emit the body
-  /// of the request. The stream may be closed to indicate a request with no
-  /// body.
+  /// all mutable fields and returns a single-subscription [ByteStream] that
+  /// emits the body of the request.
   ///
-  /// The base implementation of this returns null rather than an [InputStream];
-  /// subclasses are responsible for creating the return value. They should also
+  /// The base implementation of this returns null rather than a [ByteStream];
+  /// subclasses are responsible for creating the return value, which should be
+  /// single-subscription to ensure that no data is dropped. They should also
   /// freeze any additional mutable fields they add that don't make sense to
   /// change after the request headers are sent.
-  InputStream finalize() {
+  ByteStream finalize() {
     // TODO(nweiz): freeze headers
     if (finalized) throw new StateError("Can't finalize a finalized Request.");
     _finalized = true;
     return null;
   }
 
   /// Sends this request.
   ///
   /// This automatically initializes a new [Client] and closes that client once
   /// the request is complete. If you're planning on making multiple requests to
   /// the same server, you should use a single [Client] for all of those
   /// requests.
   Future<StreamedResponse> send() {
     var client = new Client();
     return client.send(this).then((response) {
-      // TODO(nweiz): This makes me sick to my stomach, but it's currently the
-      // best way to listen for the response stream being closed. Kill it with
-      // fire once issue 4202 is fixed.
-      new Timer.repeating(100, (timer) {
-        if (response.stream.closed) {
-          client.close();
-          timer.cancel();
-        }
-      });
-
-      return response;
-    }).catchError((_) { client.close(); });
+      var stream = onDone(response.stream, client.close);
+      return new StreamedResponse(
+          new ByteStream(stream),
+          response.statusCode,
+          response.contentLength,
+          request: response.request,
+          headers: response.headers,
+          isRedirect: response.isRedirect,
+          persistentConnection: response.persistentConnection,
+          reasonPhrase: response.reasonPhrase);
+    }).catchError((e) {
+      client.close();
+      throw e;
+    });
   }
 
   /// Throws an error if this request has been finalized.
   void _checkFinalized() {
     if (!finalized) return;
     throw new StateError("Can't modify a finalized Request.");
   }
 
   String toString() => "$method $url";
 }