[flutter_tool] Stream artifact downloads to files (#44360)

packages/flutter_tools/lib/src/base/net.dart

@@ -5,8 +5,10 @@
 import 'dart:async';
 
 import '../base/context.dart';
+import '../convert.dart';
 import '../globals.dart';
 import 'common.dart';
+import 'file_system.dart';
 import 'io.dart';
 import 'platform.dart';
 
@@ -14,16 +16,39 @@ const int kNetworkProblemExitCode = 50;
 
 typedef HttpClientFactory = HttpClient Function();
 
-/// Download a file from the given URL and return the bytes.
-Future<List<int>> fetchUrl(Uri url, {int maxAttempts}) async {
+/// Download a file from the given URL.
+///
+/// If a destination file is not provided, returns the bytes.
+///
+/// If a destination file is provided, streams the bytes to that file and
+/// returns an empty list.
+///
+/// If [maxAttempts] is exceeded, returns null.
+Future<List<int>> fetchUrl(Uri url, {
+  int maxAttempts,
+  File destFile,
+}) async {
   int attempts = 0;
   int durationSeconds = 1;
   while (true) {
     attempts += 1;
-    final List<int> result = await _attempt(url);
-    if (result != null) {
-      return result;
+    _MemoryIOSink memorySink;
+    IOSink sink;
+    if (destFile == null) {
+      memorySink = _MemoryIOSink();
+      sink = memorySink;
+    } else {
+      sink = destFile.openWrite();
+    }
+
+    final bool result = await _attempt(
+      url,
+      destSink: sink,
+    );
+    if (result) {
+      return memorySink?.writes?.takeBytes() ?? <int>[];
     }
+
     if (maxAttempts != null && attempts >= maxAttempts) {
       printStatus('Download failed -- retry $attempts');
       return null;
@@ -38,10 +63,14 @@ Future<List<int>> fetchUrl(Uri url, {int maxAttempts}) async {
 }
 
 /// Check if the given URL points to a valid endpoint.
-Future<bool> doesRemoteFileExist(Uri url) async =>
-  (await _attempt(url, onlyHeaders: true)) != null;
+Future<bool> doesRemoteFileExist(Uri url) async => await _attempt(url, onlyHeaders: true);
 
-Future<List<int>> _attempt(Uri url, { bool onlyHeaders = false }) async {
+// Returns true on success and false on failure.
+Future<bool> _attempt(Uri url, {
+  IOSink destSink,
+  bool onlyHeaders = false,
+}) async {
+  assert(onlyHeaders || destSink != null);
   printTrace('Downloading: $url');
   HttpClient httpClient;
   if (context.get<HttpClientFactory>() != null) {
@@ -81,19 +110,19 @@ Future<List<int>> _attempt(Uri url, { bool onlyHeaders = false }) async {
     );
   } on SocketException catch (error) {
     printTrace('Download error: $error');
-    return null;
+    return false;
   } on HttpException catch (error) {
     printTrace('Download error: $error');
-    return null;
+    return false;
   }
   assert(response != null);
 
   // If we're making a HEAD request, we're only checking to see if the URL is
   // valid.
   if (onlyHeaders) {
-    return (response.statusCode == 200) ? <int>[] : null;
+    return response.statusCode == HttpStatus.ok;
   }
-  if (response.statusCode != 200) {
+  if (response.statusCode != HttpStatus.ok) {
     if (response.statusCode > 0 && response.statusCode < 500) {
       throwToolExit(
         'Download failed.\n'
@@ -104,15 +133,79 @@ Future<List<int>> _attempt(Uri url, { bool onlyHeaders = false }) async {
     }
     // 5xx errors are server errors and we can try again
     printTrace('Download error: ${response.statusCode} ${response.reasonPhrase}');
-    return null;
+    return false;
   }
   printTrace('Received response from server, collecting bytes...');
   try {
-    final BytesBuilder responseBody = BytesBuilder(copy: false);
-    await response.forEach(responseBody.add);
-    return responseBody.takeBytes();
+    assert(destSink != null);
+    await response.forEach(destSink.add);
+    return true;
   } on IOException catch (error) {
     printTrace('Download error: $error');
-    return null;
+    return false;
+  } finally {
+    await destSink?.flush();
+    await destSink?.close();
+  }
+}
+
+/// An IOSink that collects whatever is written to it.
+class _MemoryIOSink implements IOSink {
+  @override
+  Encoding encoding = utf8;
+
+  final BytesBuilder writes = BytesBuilder(copy: false);
+
+  @override
+  void add(List<int> data) {
+    writes.add(data);
+  }
+
+  @override
+  Future<void> addStream(Stream<List<int>> stream) {
+    final Completer<void> completer = Completer<void>();
+    stream.listen(add).onDone(completer.complete);
+    return completer.future;
+  }
+
+  @override
+  void writeCharCode(int charCode) {
+    add(<int>[charCode]);
+  }
+
+  @override
+  void write(Object obj) {
+    add(encoding.encode('$obj'));
   }
+
+  @override
+  void writeln([ Object obj = '' ]) {
+    add(encoding.encode('$obj\n'));
+  }
+
+  @override
+  void writeAll(Iterable<dynamic> objects, [ String separator = '' ]) {
+    bool addSeparator = false;
+    for (dynamic object in objects) {
+      if (addSeparator) {
+        write(separator);
+      }
+      write(object);
+      addSeparator = true;
+    }
+  }
+
+  @override
+  void addError(dynamic error, [ StackTrace stackTrace ]) {
+    throw UnimplementedError();
+  }
+
+  @override
+  Future<void> get done => close();
+
+  @override
+  Future<void> close() async { }
+
+  @override
+  Future<void> flush() async { }
 }

packages/flutter_tools/lib/src/cache.dart

@@ -1242,8 +1242,7 @@ String flattenNameSubdirs(Uri url) {
 /// Download a file from the given [url] and write it to [location].
 Future<void> _downloadFile(Uri url, File location) async {
   _ensureExists(location.parent);
-  final List<int> fileBytes = await fetchUrl(url);
-  location.writeAsBytesSync(fileBytes, flush: true);
+  await fetchUrl(url, destFile: location);
 }
 
 Future<bool> _doesRemoteExist(String message, Uri url) async {

packages/flutter_tools/test/general.shard/base/net_test.dart

@@ -3,8 +3,12 @@
 // found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:convert';
 import 'dart:typed_data';
 
+import 'package:file/file.dart';
+import 'package:file/memory.dart';
+import 'package:flutter_tools/src/base/file_system.dart';
 import 'package:flutter_tools/src/base/io.dart' as io;
 import 'package:flutter_tools/src/base/net.dart';
 import 'package:flutter_tools/src/base/platform.dart';
@@ -14,6 +18,36 @@ import '../../src/common.dart';
 import '../../src/context.dart';
 
 void main() {
+  group('successful fetch', () {
+    const String responseString = 'response string';
+    List<int> responseData;
+
+    setUp(() {
+      responseData = utf8.encode(responseString);
+    });
+
+    testUsingContext('fetchUrl() gets the data', () async {
+      final List<int> data = await fetchUrl(Uri.parse('http://example.invalid/'));
+      expect(data, equals(responseData));
+    }, overrides: <Type, Generator>{
+      HttpClientFactory: () => () => FakeHttpClient(200, data: responseString),
+    });
+
+    testUsingContext('fetchUrl(destFile) writes the data to a file', () async {
+      final File destFile = fs.file('dest_file')..createSync();
+      final List<int> data = await fetchUrl(
+        Uri.parse('http://example.invalid/'),
+        destFile: destFile,
+      );
+      expect(data, equals(<int>[]));
+      expect(destFile.readAsStringSync(), equals(responseString));
+    }, overrides: <Type, Generator>{
+      FileSystem: () => MemoryFileSystem(),
+      HttpClientFactory: () => () => FakeHttpClient(200, data: responseString),
+      ProcessManager: () => FakeProcessManager.any(),
+    });
+  });
+
   testUsingContext('retry from 500', () async {
     String error;
     FakeAsync().run((FakeAsync time) {
@@ -249,13 +283,14 @@ class FakeHttpClientThrowing implements io.HttpClient {
 }
 
 class FakeHttpClient implements io.HttpClient {
-  FakeHttpClient(this.statusCode);
+  FakeHttpClient(this.statusCode, { this.data });
 
   final int statusCode;
+  final String data;
 
   @override
   Future<io.HttpClientRequest> getUrl(Uri url) async {
-    return FakeHttpClientRequest(statusCode);
+    return FakeHttpClientRequest(statusCode, data: data);
   }
 
   @override
@@ -286,13 +321,14 @@ class FakeHttpClientThrowingRequest implements io.HttpClient {
 }
 
 class FakeHttpClientRequest implements io.HttpClientRequest {
-  FakeHttpClientRequest(this.statusCode);
+  FakeHttpClientRequest(this.statusCode, { this.data });
 
   final int statusCode;
+  final String data;
 
   @override
   Future<io.HttpClientResponse> close() async {
-    return FakeHttpClientResponse(statusCode);
+    return FakeHttpClientResponse(statusCode, data: data);
   }
 
   @override
@@ -318,11 +354,13 @@ class FakeHttpClientRequestThrowing implements io.HttpClientRequest {
 }
 
 class FakeHttpClientResponse implements io.HttpClientResponse {
-  FakeHttpClientResponse(this.statusCode);
+  FakeHttpClientResponse(this.statusCode, { this.data });
 
   @override
   final int statusCode;
 
+  final String data;
+
   @override
   String get reasonPhrase => '<reason phrase>';
 
@@ -333,13 +371,24 @@ class FakeHttpClientResponse implements io.HttpClientResponse {
     void onDone(),
     bool cancelOnError,
   }) {
-    return Stream<Uint8List>.fromFuture(Future<Uint8List>.error(const io.SocketException('test')))
-      .listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError);
+    if (data == null) {
+      return Stream<Uint8List>.fromFuture(Future<Uint8List>.error(
+        const io.SocketException('test'),
+      )).listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError);
+    } else {
+      return Stream<Uint8List>.fromFuture(Future<Uint8List>.value(
+        utf8.encode(data) as Uint8List,
+      )).listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError);
+    }
   }
 
   @override
-  Future<dynamic> forEach(void Function(Uint8List element) action) {
-    return Future<void>.error(const io.SocketException('test'));
+  Future<dynamic> forEach(void Function(Uint8List element) action) async {
+    if (data == null) {
+      return Future<void>.error(const io.SocketException('test'));
+    } else {
+      return Future<void>.microtask(() => action(utf8.encode(data)));
+    }
   }
 
   @override