Refactor to async/await (#328)

Use an `async*` for `MultipartRequest.finalize()`.

Refactor `.then` chains to async methods.

Add test for error forwarding in multipart request. 
This relates to a TODO that had been solved but
had not been removed.

Author: natebosch

lib/src/base_client.dart

@@ -121,11 +121,10 @@ abstract class BaseClient implements Client {
   /// For more fine-grained control over the request and response, use [send] or
   /// [get] instead.
   @override
-  Future<String> read(url, {Map<String, String> headers}) {
-    return get(url, headers: headers).then((response) {
-      _checkResponseSuccess(url, response);
-      return response.body;
-    });
+  Future<String> read(url, {Map<String, String> headers}) async {
+    final response = await get(url, headers: headers);
+    _checkResponseSuccess(url, response);
+    return response.body;
   }
 
   /// Sends an HTTP GET request with the given headers to the given URL, which
@@ -138,11 +137,10 @@ abstract class BaseClient implements Client {
   /// For more fine-grained control over the request and response, use [send] or
   /// [get] instead.
   @override
-  Future<Uint8List> readBytes(url, {Map<String, String> headers}) {
-    return get(url, headers: headers).then((response) {
-      _checkResponseSuccess(url, response);
-      return response.bodyBytes;
-    });
+  Future<Uint8List> readBytes(url, {Map<String, String> headers}) async {
+    final response = await get(url, headers: headers);
+    _checkResponseSuccess(url, response);
+    return response.bodyBytes;
   }
 
   /// Sends an HTTP request and asynchronously returns the response.

lib/src/mock_client.dart

@@ -30,42 +30,39 @@ class MockClient extends BaseClient {
   /// Creates a [MockClient] with a handler that receives [Request]s and sends
   /// [Response]s.
   MockClient(MockClientHandler fn)
-      : this._((baseRequest, bodyStream) {
-          return bodyStream.toBytes().then((bodyBytes) {
-            var request = Request(baseRequest.method, baseRequest.url)
-              ..persistentConnection = baseRequest.persistentConnection
-              ..followRedirects = baseRequest.followRedirects
-              ..maxRedirects = baseRequest.maxRedirects
-              ..headers.addAll(baseRequest.headers)
-              ..bodyBytes = bodyBytes
-              ..finalize();
+      : this._((baseRequest, bodyStream) async {
+          final bodyBytes = await bodyStream.toBytes();
+          var request = Request(baseRequest.method, baseRequest.url)
+            ..persistentConnection = baseRequest.persistentConnection
+            ..followRedirects = baseRequest.followRedirects
+            ..maxRedirects = baseRequest.maxRedirects
+            ..headers.addAll(baseRequest.headers)
+            ..bodyBytes = bodyBytes
+            ..finalize();
 
-            return fn(request);
-          }).then((response) {
-            return StreamedResponse(
-                ByteStream.fromBytes(response.bodyBytes), response.statusCode,
-                contentLength: response.contentLength,
-                request: baseRequest,
-                headers: response.headers,
-                isRedirect: response.isRedirect,
-                persistentConnection: response.persistentConnection,
-                reasonPhrase: response.reasonPhrase);
-          });
+          final response = await fn(request);
+          return StreamedResponse(
+              ByteStream.fromBytes(response.bodyBytes), response.statusCode,
+              contentLength: response.contentLength,
+              request: baseRequest,
+              headers: response.headers,
+              isRedirect: response.isRedirect,
+              persistentConnection: response.persistentConnection,
+              reasonPhrase: response.reasonPhrase);
         });
 
   /// Creates a [MockClient] with a handler that receives [StreamedRequest]s and
   /// sends [StreamedResponse]s.
   MockClient.streaming(MockClientStreamHandler fn)
-      : this._((request, bodyStream) {
-          return fn(request, bodyStream).then((response) {
-            return StreamedResponse(response.stream, response.statusCode,
-                contentLength: response.contentLength,
-                request: request,
-                headers: response.headers,
-                isRedirect: response.isRedirect,
-                persistentConnection: response.persistentConnection,
-                reasonPhrase: response.reasonPhrase);
-          });
+      : this._((request, bodyStream) async {
+          final response = await fn(request, bodyStream);
+          return StreamedResponse(response.stream, response.statusCode,
+              contentLength: response.contentLength,
+              request: request,
+              headers: response.headers,
+              isRedirect: response.isRedirect,
+              persistentConnection: response.persistentConnection,
+              reasonPhrase: response.reasonPhrase);
         });
 
   @override

lib/src/multipart_request.dart

@@ -87,40 +87,31 @@ class MultipartRequest extends BaseRequest {
   /// Freezes all mutable fields and returns a single-subscription [ByteStream]
   /// that will emit the request body.
   @override
-  ByteStream finalize() {
+  ByteStream finalize() => ByteStream(_finalize());
+
+  Stream<List<int>> _finalize() async* {
     // TODO(nweiz): freeze fields and files
     var boundary = _boundaryString();
     headers['content-type'] = 'multipart/form-data; boundary=$boundary';
     super.finalize();
-
-    var controller = StreamController<List<int>>(sync: true);
-
-    void writeAscii(String string) {
-      controller.add(utf8.encode(string));
+    const line = [13, 10]; // \r\n
+    final separator = utf8.encode('--$boundary\r\n');
+    final close = utf8.encode('--$boundary--\r\n');
+
+    for (var field in fields.entries) {
+      yield separator;
+      yield utf8.encode(_headerForField(field.key, field.value));
+      yield utf8.encode(field.value);
+      yield line;
     }
 
-    writeUtf8(String string) => controller.add(utf8.encode(string));
-    writeLine() => controller.add([13, 10]); // \r\n
-
-    fields.forEach((name, value) {
-      writeAscii('--$boundary\r\n');
-      writeAscii(_headerForField(name, value));
-      writeUtf8(value);
-      writeLine();
-    });
-
-    Future.forEach(_files, (MultipartFile file) {
-      writeAscii('--$boundary\r\n');
-      writeAscii(_headerForFile(file));
-      return controller.addStream(file.finalize()).then((_) => writeLine());
-    }).then((_) {
-      // TODO(nweiz): pass any errors propagated through this future on to
-      // the stream. See issue 3657.
-      writeAscii('--$boundary--\r\n');
-      controller.close();
-    });
-
-    return ByteStream(controller.stream);
+    for (final file in _files) {
+      yield separator;
+      yield utf8.encode(_headerForFile(file));
+      yield* file.finalize();
+      yield line;
+    }
+    yield close;
   }
 
   /// Returns the header string for a field.

lib/src/response.dart

@@ -60,15 +60,14 @@ class Response extends BaseResponse {
 
   /// Creates a new HTTP response by waiting for the full body to become
   /// available from a [StreamedResponse].
-  static Future<Response> fromStream(StreamedResponse response) {
-    return response.stream.toBytes().then((body) {
-      return Response.bytes(body, response.statusCode,
-          request: response.request,
-          headers: response.headers,
-          isRedirect: response.isRedirect,
-          persistentConnection: response.persistentConnection,
-          reasonPhrase: response.reasonPhrase);
-    });
+  static Future<Response> fromStream(StreamedResponse response) async {
+    final body = await response.stream.toBytes();
+    return Response.bytes(body, response.statusCode,
+        request: response.request,
+        headers: response.headers,
+        isRedirect: response.isRedirect,
+        persistentConnection: response.persistentConnection,
+        reasonPhrase: response.reasonPhrase);
   }
 }
 

test/multipart_test.dart

@@ -239,4 +239,11 @@ void main() {
         --{{boundary}}--
         '''));
   });
+
+  test('with a file that has an error', () async {
+    var file = http.MultipartFile(
+        'file', Future<List<int>>.error('error').asStream(), 1);
+    var request = http.MultipartRequest('POST', dummyUrl)..files.add(file);
+    expect(request.finalize().drain(), throwsA('error'));
+  });
 }