Use separate isolate for image loading. (#34188)

* Use separate isolate for image loading. Use TransferableTypedData to const-cost receive bytes from that isolate.

packages/flutter/lib/src/foundation/consolidate_response.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
+import 'dart:isolate';
 import 'dart:typed_data';
 
 /// Signature for getting notified when chunks of bytes are received while
@@ -22,11 +23,11 @@ import 'dart:typed_data';
 /// returned to the client and the total size of the response may not be known
 /// until the request has been fully processed).
 ///
-/// This is used in [consolidateHttpClientResponseBytes].
+/// This is used in [getHttpClientResponseBytes].
 typedef BytesReceivedCallback = void Function(int cumulative, int total);
 
 /// Efficiently converts the response body of an [HttpClientResponse] into a
-/// [Uint8List].
+/// [TransferableTypedData].
 ///
 /// The future returned will forward any error emitted by `response`.
 ///
@@ -43,13 +44,13 @@ typedef BytesReceivedCallback = void Function(int cumulative, int total);
 /// bytes from this method (assuming the response is sending compressed bytes),
 /// set both [HttpClient.autoUncompress] to false and the `autoUncompress`
 /// parameter to false.
-Future<Uint8List> consolidateHttpClientResponseBytes(
+Future<TransferableTypedData> getHttpClientResponseBytes(
   HttpClientResponse response, {
   bool autoUncompress = true,
   BytesReceivedCallback onBytesReceived,
 }) {
   assert(autoUncompress != null);
-  final Completer<Uint8List> completer = Completer<Uint8List>.sync();
+  final Completer<TransferableTypedData> completer = Completer<TransferableTypedData>.sync();
 
   final _OutputBuffer output = _OutputBuffer();
   ByteConversionSink sink = output;
@@ -89,41 +90,54 @@ Future<Uint8List> consolidateHttpClientResponseBytes(
     }
   }, onDone: () {
     sink.close();
-    completer.complete(output.bytes);
+    completer.complete(TransferableTypedData.fromList(output.chunks));
   }, onError: completer.completeError, cancelOnError: true);
 
   return completer.future;
 }
 
+/// Efficiently converts the response body of an [HttpClientResponse] into a
+/// [Uint8List].
+///
+/// (This method is deprecated - use [getHttpClientResponseBytes] instead.)
+///
+/// The future returned will forward any error emitted by `response`.
+///
+/// The `onBytesReceived` callback, if specified, will be invoked for every
+/// chunk of bytes that is received while consolidating the response bytes.
+/// If the callback throws an error, processing of the response will halt, and
+/// the returned future will complete with the error that was thrown by the
+/// callback. For more information on how to interpret the parameters to the
+/// callback, see the documentation on [BytesReceivedCallback].
+///
+/// If the `response` is gzipped and the `autoUncompress` parameter is true,
+/// this will automatically un-compress the bytes in the returned list if it
+/// hasn't already been done via [HttpClient.autoUncompress]. To get compressed
+/// bytes from this method (assuming the response is sending compressed bytes),
+/// set both [HttpClient.autoUncompress] to false and the `autoUncompress`
+/// parameter to false.
+@Deprecated('Use getHttpClientResponseBytes instead')
+Future<Uint8List> consolidateHttpClientResponseBytes(
+  HttpClientResponse response, {
+  bool autoUncompress = true,
+  BytesReceivedCallback onBytesReceived,
+}) async {
+  final TransferableTypedData bytes = await getHttpClientResponseBytes(
+    response,
+    autoUncompress: autoUncompress,
+    onBytesReceived: onBytesReceived,
+  );
+  return bytes.materialize().asUint8List();
+}
+
 class _OutputBuffer extends ByteConversionSinkBase {
-  List<List<int>> _chunks = <List<int>>[];
-  int _contentLength = 0;
-  Uint8List _bytes;
+  final List<Uint8List> chunks = <Uint8List>[];
 
   @override
   void add(List<int> chunk) {
-    assert(_bytes == null);
-    _chunks.add(chunk);
-    _contentLength += chunk.length;
+    chunks.add(chunk);
   }
 
   @override
-  void close() {
-    if (_bytes != null) {
-      // We've already been closed; this is a no-op
-      return;
-    }
-    _bytes = Uint8List(_contentLength);
-    int offset = 0;
-    for (List<int> chunk in _chunks) {
-      _bytes.setRange(offset, offset + chunk.length, chunk);
-      offset += chunk.length;
-    }
-    _chunks = null;
-  }
-
-  Uint8List get bytes {
-    assert(_bytes != null);
-    return _bytes;
-  }
+  void close() {}
 }

packages/flutter/lib/src/painting/_network_image_io.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 import 'dart:io';
+import 'dart:isolate';
 import 'dart:typed_data';
 import 'dart:ui' as ui;
 
@@ -14,7 +15,7 @@ import 'debug.dart';
 import 'image_provider.dart' as image_provider;
 import 'image_stream.dart';
 
-/// The dart:io implemenation of [image_provider.NetworkImage].
+/// The dart:io implementation of [image_provider.NetworkImage].
 class NetworkImage extends image_provider.ImageProvider<image_provider.NetworkImage> implements image_provider.NetworkImage {
   /// Creates an object that fetches the image at the given URL.
   ///
@@ -39,9 +40,9 @@ class NetworkImage extends image_provider.ImageProvider<image_provider.NetworkIm
 
   @override
   ImageStreamCompleter load(image_provider.NetworkImage key) {
-    // Ownership of this controller is handed off to [_loadAsync]; it is that
-    // method's responsibility to close the controller's stream when the image
-    // has been loaded or an error is thrown.
+    // Ownership of this controller is handed off to [_loadAsync];
+    // it is that method's responsibility to close the controller's stream when
+    // the image has been loaded or an error is thrown.
     final StreamController<ImageChunkEvent> chunkEvents = StreamController<ImageChunkEvent>();
 
     return MultiFrameImageStreamCompleter(
@@ -57,54 +58,130 @@ class NetworkImage extends image_provider.ImageProvider<image_provider.NetworkIm
     );
   }
 
-  // Do not access this field directly; use [_httpClient] instead.
-  // We set `autoUncompress` to false to ensure that we can trust the value of
-  // the `Content-Length` HTTP header. We automatically uncompress the content
-  // in our call to [consolidateHttpClientResponseBytes].
-  static final HttpClient _sharedHttpClient = HttpClient()..autoUncompress = false;
-
-  static HttpClient get _httpClient {
-    HttpClient client = _sharedHttpClient;
-    assert(() {
-      if (debugNetworkImageHttpClientProvider != null)
-        client = debugNetworkImageHttpClientProvider();
-      return true;
-    }());
-    return client;
-  }
+  // For [_pendingLoader] we don't need the value(worker isolate), just the future
+  // itself that is used as an indicator that successive load requests should be
+  // added to the list of pending load requests [_pendingLoadRequests].
+  static Future<void> _pendingLoader;
+  static RawReceivePort _loaderErrorHandler;
+  static List<_DownloadRequest> _pendingLoadRequests;
+  static SendPort _requestPort;
 
   Future<ui.Codec> _loadAsync(
     NetworkImage key,
     StreamController<ImageChunkEvent> chunkEvents,
   ) async {
+    RawReceivePort downloadResponseHandler;
     try {
       assert(key == this);
 
       final Uri resolved = Uri.base.resolve(key.url);
-      final HttpClientRequest request = await _httpClient.getUrl(resolved);
-      headers?.forEach((String name, String value) {
-        request.headers.add(name, value);
+
+      final Completer<TransferableTypedData> bytesCompleter = Completer<TransferableTypedData>();
+      downloadResponseHandler = RawReceivePort((_DownloadResponse response) {
+        if (response.bytes != null) {
+          if (bytesCompleter.isCompleted) {
+            // If an uncaught error occurred in the worker isolate, we'll have
+            // already completed our bytes completer.
+            return;
+          }
+          bytesCompleter.complete(response.bytes);
+        } else if (response.chunkEvent != null) {
+          chunkEvents.add(response.chunkEvent);
+        } else if (response.error != null) {
+          bytesCompleter.completeError(response.error);
+        } else {
+          assert(false);
+        }
       });
-      final HttpClientResponse response = await request.close();
-      if (response.statusCode != HttpStatus.ok)
-        throw image_provider.NetworkImageLoadException(statusCode: response.statusCode, uri: resolved);
 
-      final Uint8List bytes = await consolidateHttpClientResponseBytes(
-        response,
-        onBytesReceived: (int cumulative, int total) {
-          chunkEvents.add(ImageChunkEvent(
-            cumulativeBytesLoaded: cumulative,
-            expectedTotalBytes: total,
-          ));
-        },
+      // This will keep references to [debugNetworkImageHttpClientProvider] tree-shaken
+      // out of release builds.
+      HttpClientProvider httpClientProvider;
+      assert(() { httpClientProvider = debugNetworkImageHttpClientProvider; return true; }());
+
+      final _DownloadRequest downloadRequest = _DownloadRequest(
+        downloadResponseHandler.sendPort,
+        resolved,
+        headers,
+        httpClientProvider,
       );
-      if (bytes.lengthInBytes == 0)
+      if (_requestPort != null) {
+        // If worker isolate is properly set up ([_requestPort] is holding
+        // initialized [SendPort]), then just send download request down to it.
+        _requestPort.send(downloadRequest);
+      } else {
+        if (_pendingLoader == null) {
+          // If worker isolate creation was not started, start creation now.
+          _spawnAndSetupIsolate();
+        }
+        // Record download request so it can either send a request when isolate is ready or handle errors.
+        _pendingLoadRequests.add(downloadRequest);
+      }
+
+      final TransferableTypedData transferable = await bytesCompleter.future;
+
+      final Uint8List bytes = transferable.materialize().asUint8List();
+      if (bytes.isEmpty)
         throw Exception('NetworkImage is an empty file: $resolved');
 
       return PaintingBinding.instance.instantiateImageCodec(bytes);
     } finally {
       chunkEvents.close();
+      downloadResponseHandler?.close();
+    }
+  }
+
+  void _spawnAndSetupIsolate() {
+    assert(_pendingLoadRequests == null);
+    assert(_loaderErrorHandler == null);
+    assert(_pendingLoader == null);
+    _pendingLoadRequests = <_DownloadRequest>[];
+    _pendingLoader = _spawnIsolate()..then((Isolate isolate) {
+      _loaderErrorHandler = RawReceivePort((List<dynamic> errorAndStackTrace) {
+        _cleanupDueToError(errorAndStackTrace[0]);
+      });
+      isolate.addErrorListener(_loaderErrorHandler.sendPort);
+      isolate.resume(isolate.pauseCapability);
+    }).catchError((dynamic error, StackTrace stackTrace) {
+      _cleanupDueToError(error);
+    });
+  }
+
+  void _cleanupDueToError(dynamic error) {
+    for (_DownloadRequest request in _pendingLoadRequests) {
+      request.handleError(error);
     }
+    _pendingLoadRequests = null;
+    _pendingLoader = null;
+    _loaderErrorHandler.close();
+    _loaderErrorHandler = null;
+  }
+
+  Future<Isolate> _spawnIsolate() {
+    // Once worker isolate is up and running it sends it's [sendPort] over
+    // [communicationBootstrapHandler] receive port.
+    // If [sendPort] is [null], it indicates that worker isolate exited after
+    // being idle.
+    final RawReceivePort communicationBootstrapHandler = RawReceivePort((SendPort sendPort) {
+      _requestPort = sendPort;
+      if (sendPort == null) {
+        assert(_pendingLoadRequests.isEmpty);
+        _pendingLoader = null;
+        _pendingLoadRequests = null;
+        _loaderErrorHandler.close();
+        _loaderErrorHandler = null;
+        return;
+      }
+
+      // When we received [SendPort] for the worker isolate, we send all
+      // pending requests that were accumulated before worker isolate provided
+      // it's port (before [_requestPort] was populated).
+      _pendingLoadRequests.forEach(sendPort.send);
+      _pendingLoadRequests.clear();
+    });
+
+    return Isolate.spawn<SendPort>(_initializeWorkerIsolate,
+      communicationBootstrapHandler.sendPort, paused: true);
   }
 
   @override
@@ -112,8 +189,7 @@ class NetworkImage extends image_provider.ImageProvider<image_provider.NetworkIm
     if (other.runtimeType != runtimeType)
       return false;
     final NetworkImage typedOther = other;
-    return url == typedOther.url
-        && scale == typedOther.scale;
+    return url == typedOther.url && scale == typedOther.scale;
   }
 
   @override
@@ -122,3 +198,95 @@ class NetworkImage extends image_provider.ImageProvider<image_provider.NetworkIm
   @override
   String toString() => '$runtimeType("$url", scale: $scale)';
 }
+
+@immutable
+class _DownloadResponse {
+  const _DownloadResponse.bytes(this.bytes) : assert(bytes != null), chunkEvent = null, error = null;
+  const _DownloadResponse.chunkEvent(this.chunkEvent) : assert(chunkEvent != null), bytes = null, error = null;
+  const _DownloadResponse.error(this.error) : assert(error != null), bytes = null, chunkEvent = null;
+
+  final TransferableTypedData bytes;
+  final ImageChunkEvent chunkEvent;
+  final dynamic error;
+}
+
+@immutable
+class _DownloadRequest {
+  const _DownloadRequest(this.sendPort, this.uri, this.headers, this.httpClientProvider) :
+      assert(sendPort != null), assert(uri != null);
+
+  final SendPort sendPort;
+  final Uri uri;
+  final Map<String, String> headers;
+  final HttpClientProvider httpClientProvider;
+
+  void handleError(dynamic error) { sendPort.send(_DownloadResponse.error(error)); }
+}
+
+// We set `autoUncompress` to false to ensure that we can trust the value of
+// the `Content-Length` HTTP header. We automatically uncompress the content
+// in our call to [getHttpClientResponseBytes].
+final HttpClient _sharedHttpClient = HttpClient()..autoUncompress = false;
+const Duration _idleDuration = Duration(seconds: 60);
+
+/// Sets up the worker isolate to listen for incoming [_DownloadRequest]s from
+/// the main isolate.
+///
+/// This method runs on a worker isolate.
+///
+/// The `handshakeSendPort` argument is this worker isolate's communications
+/// link back to the main isolate. It is used to set-up the channel with which
+/// the main isolate sends download requests to the worker isolate.
+void _initializeWorkerIsolate(SendPort handshakeSendPort) {
+  int ongoingRequests = 0;
+  Timer idleTimer;
+  RawReceivePort downloadRequestHandler;
+
+  // Sets up a handler that processes download requests messages.
+  downloadRequestHandler = RawReceivePort((_DownloadRequest downloadRequest) async {
+    ongoingRequests++;
+    idleTimer?.cancel();
+    final HttpClient httpClient = downloadRequest.httpClientProvider != null
+        ? downloadRequest.httpClientProvider()
+        : _sharedHttpClient;
+
+    try {
+      final HttpClientRequest request = await httpClient.getUrl(downloadRequest.uri);
+      downloadRequest.headers?.forEach((String name, String value) {
+        request.headers.add(name, value);
+      });
+      final HttpClientResponse response = await request.close();
+      if (response.statusCode != HttpStatus.ok) {
+        throw image_provider.NetworkImageLoadException(
+          statusCode: response?.statusCode,
+          uri: downloadRequest.uri,
+        );
+      }
+      final TransferableTypedData transferable = await getHttpClientResponseBytes(
+        response,
+        onBytesReceived: (int cumulative, int total) {
+          downloadRequest.sendPort.send(_DownloadResponse.chunkEvent(
+            ImageChunkEvent(
+              cumulativeBytesLoaded: cumulative,
+              expectedTotalBytes: total,
+            ),
+          ));
+        },
+      );
+      downloadRequest.sendPort.send(_DownloadResponse.bytes(transferable));
+    } catch (error) {
+      downloadRequest.sendPort.send(_DownloadResponse.error(error));
+    }
+    ongoingRequests--;
+    if (ongoingRequests == 0) {
+      idleTimer = Timer(_idleDuration, () {
+        assert(ongoingRequests == 0);
+        // [null] indicates that worker is going down.
+        handshakeSendPort.send(null);
+        downloadRequestHandler.close();
+      });
+    }
+  });
+
+  handshakeSendPort.send(downloadRequestHandler.sendPort);
+}

packages/flutter/lib/src/services/asset_bundle.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
+import 'dart:isolate';
 import 'dart:typed_data';
 
 import 'package:flutter/foundation.dart';
@@ -120,8 +121,8 @@ class NetworkAssetBundle extends AssetBundle {
         'Unable to load asset: $key\n'
         'HTTP status code: ${response.statusCode}'
       );
-    final Uint8List bytes = await consolidateHttpClientResponseBytes(response);
-    return bytes.buffer.asByteData();
+    final TransferableTypedData transferable = await getHttpClientResponseBytes(response);
+    return transferable.materialize().asByteData();
   }
 
   /// Retrieve a string from the asset bundle, parse it with the given function,

packages/flutter/test/foundation/consolidate_response_test.dart

@@ -6,6 +6,7 @@
 
 import 'dart:async';
 import 'dart:io';
+import 'dart:isolate';
 import 'dart:typed_data';
 
 import 'package:flutter/foundation.dart';
@@ -14,7 +15,7 @@ import 'package:mockito/mockito.dart';
 import '../flutter_test_alternative.dart';
 
 void main() {
-  group(consolidateHttpClientResponseBytes, () {
+  group(getHttpClientResponseBytes, () {
     final Uint8List chunkOne = Uint8List.fromList(<int>[0, 1, 2, 3, 4, 5]);
     final Uint8List chunkTwo = Uint8List.fromList(<int>[6, 7, 8, 9, 10]);
     MockHttpClientResponse response;
@@ -46,24 +47,24 @@ void main() {
     test('Converts an HttpClientResponse with contentLength to bytes', () async {
       when(response.contentLength)
           .thenReturn(chunkOne.length + chunkTwo.length);
-      final List<int> bytes =
-          await consolidateHttpClientResponseBytes(response);
+      final List<int> bytes = (await getHttpClientResponseBytes(response))
+          .materialize().asUint8List();
 
       expect(bytes, <int>[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
     });
 
     test('Converts a compressed HttpClientResponse with contentLength to bytes', () async {
       when(response.contentLength).thenReturn(chunkOne.length);
-      final List<int> bytes =
-          await consolidateHttpClientResponseBytes(response);
+      final List<int> bytes = (await getHttpClientResponseBytes(response))
+          .materialize().asUint8List();
 
       expect(bytes, <int>[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
     });
 
     test('Converts an HttpClientResponse without contentLength to bytes', () async {
       when(response.contentLength).thenReturn(-1);
-      final List<int> bytes =
-          await consolidateHttpClientResponseBytes(response);
+      final List<int> bytes = (await getHttpClientResponseBytes(response))
+          .materialize().asUint8List();
 
       expect(bytes, <int>[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
     });
@@ -72,7 +73,7 @@ void main() {
       final int syntheticTotal = (chunkOne.length + chunkTwo.length) * 2;
       when(response.contentLength).thenReturn(syntheticTotal);
       final List<int> records = <int>[];
-      await consolidateHttpClientResponseBytes(
+      await getHttpClientResponseBytes(
         response,
         onBytesReceived: (int cumulative, int total) {
           records.addAll(<int>[cumulative, total]);
@@ -110,13 +111,13 @@ void main() {
       });
       when(response.contentLength).thenReturn(-1);
 
-      expect(consolidateHttpClientResponseBytes(response),
+      expect(getHttpClientResponseBytes(response),
           throwsA(isInstanceOf<Exception>()));
     });
 
     test('Propagates error to Future return value if onBytesReceived throws', () async {
       when(response.contentLength).thenReturn(-1);
-      final Future<List<int>> result = consolidateHttpClientResponseBytes(
+      final Future<TransferableTypedData> result = getHttpClientResponseBytes(
         response,
         onBytesReceived: (int cumulative, int total) {
           throw 'misbehaving callback';
@@ -157,14 +158,14 @@ void main() {
       test('Uncompresses GZIP bytes if autoUncompress is true and response.compressionState is compressed', () async {
         when(response.compressionState).thenReturn(HttpClientResponseCompressionState.compressed);
         when(response.contentLength).thenReturn(gzipped.length);
-        final List<int> bytes = await consolidateHttpClientResponseBytes(response);
+        final List<int> bytes = (await getHttpClientResponseBytes(response)).materialize().asUint8List();
         expect(bytes, <int>[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
       });
 
       test('returns gzipped bytes if autoUncompress is false and response.compressionState is compressed', () async {
         when(response.compressionState).thenReturn(HttpClientResponseCompressionState.compressed);
         when(response.contentLength).thenReturn(gzipped.length);
-        final List<int> bytes = await consolidateHttpClientResponseBytes(response, autoUncompress: false);
+        final List<int> bytes = (await getHttpClientResponseBytes(response, autoUncompress: false)).materialize().asUint8List();
         expect(bytes, gzipped);
       });
 
@@ -172,7 +173,7 @@ void main() {
         when(response.compressionState).thenReturn(HttpClientResponseCompressionState.compressed);
         when(response.contentLength).thenReturn(gzipped.length);
         final List<int> records = <int>[];
-        await consolidateHttpClientResponseBytes(
+        await getHttpClientResponseBytes(
           response,
           onBytesReceived: (int cumulative, int total) {
             records.addAll(<int>[cumulative, total]);
@@ -192,7 +193,7 @@ void main() {
         when(response.compressionState).thenReturn(HttpClientResponseCompressionState.decompressed);
         when(response.contentLength).thenReturn(syntheticTotal);
         final List<int> records = <int>[];
-        await consolidateHttpClientResponseBytes(
+        await getHttpClientResponseBytes(
           response,
           onBytesReceived: (int cumulative, int total) {
             records.addAll(<int>[cumulative, total]);

packages/flutter/test/painting/image_provider_test.dart

@@ -144,26 +144,11 @@ void main() {
     });
 
     group(NetworkImage, () {
-      MockHttpClient httpClient;
-
-      setUp(() {
-        httpClient = MockHttpClient();
-        debugNetworkImageHttpClientProvider = () => httpClient;
-      });
-
-      tearDown(() {
-        debugNetworkImageHttpClientProvider = null;
-      });
-
       test('Expect thrown exception with statusCode', () async {
         final int errorStatusCode = HttpStatus.notFound;
         const String requestUrl = 'foo-url';
 
-        final MockHttpClientRequest request = MockHttpClientRequest();
-        final MockHttpClientResponse response = MockHttpClientResponse();
-        when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
-        when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
-        when(response.statusCode).thenReturn(errorStatusCode);
+        debugNetworkImageHttpClientProvider = returnErrorStatusCode;
 
         final Completer<dynamic> caughtError = Completer<dynamic>();
 
@@ -188,32 +173,40 @@ void main() {
       });
 
       test('Uses the HttpClient provided by debugNetworkImageHttpClientProvider if set', () async {
-        when(httpClient.getUrl(any)).thenThrow('client1');
+        debugNetworkImageHttpClientProvider = throwOnAnyClient1;
+
         final List<dynamic> capturedErrors = <dynamic>[];
 
         Future<void> loadNetworkImage() async {
           final NetworkImage networkImage = NetworkImage(nonconst('foo'));
-          final ImageStreamCompleter completer = networkImage.load(networkImage);
-          completer.addListener(ImageStreamListener(
-            (ImageInfo image, bool synchronousCall) { },
+          final Completer<bool> completer = Completer<bool>();
+          networkImage.load(networkImage).addListener(ImageStreamListener(
+            (ImageInfo image, bool synchronousCall) {
+              completer.complete(true);
+            },
             onError: (dynamic error, StackTrace stackTrace) {
               capturedErrors.add(error);
+              completer.complete(false);
             },
           ));
-          await Future<void>.value();
+          await completer.future;
         }
 
         await loadNetworkImage();
-        expect(capturedErrors, <dynamic>['client1']);
-        final MockHttpClient client2 = MockHttpClient();
-        when(client2.getUrl(any)).thenThrow('client2');
-        debugNetworkImageHttpClientProvider = () => client2;
+        expect(capturedErrors, isNotNull);
+        expect(capturedErrors.length, 1);
+        expect(capturedErrors[0], equals('client1'));
+
+        debugNetworkImageHttpClientProvider = throwOnAnyClient2;
         await loadNetworkImage();
-        expect(capturedErrors, <dynamic>['client1', 'client2']);
+        expect(capturedErrors, isNotNull);
+        expect(capturedErrors.length, 2);
+        expect(capturedErrors[0], equals('client1'));
+        expect(capturedErrors[1], equals('client2'));
       }, skip: isBrowser);
 
       test('Propagates http client errors during resolve()', () async {
-        when(httpClient.getUrl(any)).thenThrow(Error());
+        debugNetworkImageHttpClientProvider = throwErrorOnAny;
         bool uncaught = false;
 
         await runZoned(() async {
@@ -237,40 +230,16 @@ void main() {
       });
 
       test('Notifies listeners of chunk events', () async {
+        debugNetworkImageHttpClientProvider = respondOnAny;
+
         const int chunkSize = 8;
-        final List<Uint8List> chunks = <Uint8List>[
-          for (int offset = 0; offset < kTransparentImage.length; offset += chunkSize)
-            Uint8List.fromList(kTransparentImage.skip(offset).take(chunkSize).toList()),
-        ];
-        final Completer<void> imageAvailable = Completer<void>();
-        final MockHttpClientRequest request = MockHttpClientRequest();
-        final MockHttpClientResponse response = MockHttpClientResponse();
-        when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
-        when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
-        when(response.statusCode).thenReturn(HttpStatus.ok);
-        when(response.contentLength).thenReturn(kTransparentImage.length);
-        when(response.listen(
-          any,
-          onDone: anyNamed('onDone'),
-          onError: anyNamed('onError'),
-          cancelOnError: anyNamed('cancelOnError'),
-        )).thenAnswer((Invocation invocation) {
-          final void Function(List<int>) onData = invocation.positionalArguments[0];
-          final void Function(Object) onError = invocation.namedArguments[#onError];
-          final void Function() onDone = invocation.namedArguments[#onDone];
-          final bool cancelOnError = invocation.namedArguments[#cancelOnError];
-
-          return Stream<Uint8List>.fromIterable(chunks).listen(
-            onData,
-            onDone: onDone,
-            onError: onError,
-            cancelOnError: cancelOnError,
-          );
-        });
+        final List<Uint8List> chunks = createChunks(chunkSize);
 
+        final Completer<void> imageAvailable = Completer<void>();
         final ImageProvider imageProvider = NetworkImage(nonconst('foo'));
         final ImageStream result = imageProvider.resolve(ImageConfiguration.empty);
         final List<ImageChunkEvent> events = <ImageChunkEvent>[];
+
         result.addListener(ImageStreamListener(
           (ImageInfo image, bool synchronousCall) {
             imageAvailable.complete();
@@ -289,6 +258,69 @@ void main() {
           expect(events[i].expectedTotalBytes, kTransparentImage.length);
         }
       }, skip: isBrowser);
+
+      test('Uses http request headers', () async {
+        debugNetworkImageHttpClientProvider = respondOnAnyWithHeaders;
+
+        final Completer<bool> imageAvailable = Completer<bool>();
+        final ImageProvider imageProvider = NetworkImage(nonconst('foo'),
+          headers: const <String, String>{'flutter': 'flutter'},
+        );
+        final ImageStream result = imageProvider.resolve(ImageConfiguration.empty);
+        result.addListener(ImageStreamListener(
+              (ImageInfo image, bool synchronousCall) {
+            imageAvailable.complete(true);
+          },
+          onError: (dynamic error, StackTrace stackTrace) {
+            imageAvailable.completeError(error, stackTrace);
+          },
+        ));
+        expect(await imageAvailable.future, isTrue);
+      }, skip: isBrowser);
+
+      test('Handles http stream errors', () async {
+        debugNetworkImageHttpClientProvider = respondErrorOnAny;
+
+        final Completer<String> imageAvailable = Completer<String>();
+        final ImageProvider imageProvider = NetworkImage(nonconst('bar'));
+        final ImageStream result = imageProvider.resolve(ImageConfiguration.empty);
+        final List<ImageChunkEvent> events = <ImageChunkEvent>[];
+
+        result.addListener(ImageStreamListener(
+          (ImageInfo image, bool synchronousCall) {
+            imageAvailable.complete(null);
+          },
+          onChunk: (ImageChunkEvent event) {
+            events.add(event);
+          },
+          onError: (dynamic error, StackTrace stackTrace) {
+            imageAvailable.complete(error);
+          },
+        ));
+        final String error = await imageAvailable.future;
+        expect(error, 'failed chunk');
+      }, skip: isBrowser);
+
+      test('Handles http connection errors', () async {
+        debugNetworkImageHttpClientProvider = respondErrorOnConnection;
+
+        final Completer<dynamic> imageAvailable = Completer<dynamic>();
+        final ImageProvider imageProvider = NetworkImage(nonconst('baz'));
+        final ImageStream result = imageProvider.resolve(ImageConfiguration.empty);
+        result.addListener(ImageStreamListener(
+              (ImageInfo image, bool synchronousCall) {
+            imageAvailable.complete(null);
+          },
+          onError: (dynamic error, StackTrace stackTrace) {
+            imageAvailable.complete(error);
+          },
+        ));
+        final dynamic err = await imageAvailable.future;
+        expect(err, const TypeMatcher<NetworkImageLoadException>()
+            .having((NetworkImageLoadException e) => e.toString(), 'e', startsWith('HTTP request failed'))
+            .having((NetworkImageLoadException e) => e.statusCode, 'statusCode', HttpStatus.badGateway)
+            .having((NetworkImageLoadException e) => e.uri.toString(), 'uri', endsWith('/baz')));
+      }, skip: isBrowser);
     });
   });
 }
@@ -296,3 +328,169 @@ void main() {
 class MockHttpClient extends Mock implements HttpClient {}
 class MockHttpClientRequest extends Mock implements HttpClientRequest {}
 class MockHttpClientResponse extends Mock implements HttpClientResponse {}
+class MockHttpHeaders extends Mock implements HttpHeaders {}
+
+HttpClient returnErrorStatusCode() {
+  final int errorStatusCode = HttpStatus.notFound;
+
+  debugNetworkImageHttpClientProvider = returnErrorStatusCode;
+
+  final MockHttpClientRequest request = MockHttpClientRequest();
+  final MockHttpClientResponse response = MockHttpClientResponse();
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
+  when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
+  when(response.statusCode).thenReturn(errorStatusCode);
+
+  return httpClient;
+}
+
+HttpClient throwOnAnyClient1() {
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenThrow('client1');
+  return httpClient;
+}
+
+HttpClient throwOnAnyClient2() {
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenThrow('client2');
+  return httpClient;
+}
+
+HttpClient throwErrorOnAny() {
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenThrow(Exception());
+  return httpClient;
+}
+
+HttpClient respondOnAny() {
+  const int chunkSize = 8;
+  final List<Uint8List> chunks = createChunks(chunkSize);
+  final MockHttpClientRequest request = MockHttpClientRequest();
+  final MockHttpClientResponse response = MockHttpClientResponse();
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
+  when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
+  when(response.statusCode).thenReturn(HttpStatus.ok);
+  when(response.contentLength).thenReturn(kTransparentImage.length);
+  when(response.listen(
+    any,
+    onDone: anyNamed('onDone'),
+    onError: anyNamed('onError'),
+    cancelOnError: anyNamed('cancelOnError'),
+  )).thenAnswer((Invocation invocation) {
+    final void Function(Uint8List) onData = invocation.positionalArguments[0];
+    final void Function(Object) onError = invocation.namedArguments[#onError];
+    final void Function() onDone = invocation.namedArguments[#onDone];
+    final bool cancelOnError = invocation.namedArguments[#cancelOnError];
+
+    return Stream<Uint8List>.fromIterable(chunks).listen(
+      onData,
+      onDone: onDone,
+      onError: onError,
+      cancelOnError: cancelOnError,
+    );
+  });
+  return httpClient;
+}
+
+HttpClient respondOnAnyWithHeaders() {
+  final List<Invocation> invocations = <Invocation>[];
+
+  const int chunkSize = 8;
+  final List<Uint8List> chunks = createChunks(chunkSize);
+  final MockHttpClientRequest request = MockHttpClientRequest();
+  final MockHttpClientResponse response = MockHttpClientResponse();
+  final MockHttpClient httpClient = MockHttpClient();
+  final MockHttpHeaders headers = MockHttpHeaders();
+  when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
+  when(request.headers).thenReturn(headers);
+  when(headers.add(any, any)).thenAnswer((Invocation invocation) {
+    invocations.add(invocation);
+  });
+
+  when(request.close()).thenAnswer((Invocation invocation) {
+    if (invocations.length == 1 &&
+        invocations[0].positionalArguments.length == 2 &&
+        invocations[0].positionalArguments[0] == 'flutter' &&
+        invocations[0].positionalArguments[1] == 'flutter') {
+      return Future<HttpClientResponse>.value(response);
+    } else {
+      return Future<HttpClientResponse>.value(null);
+    }
+  });
+  when(response.statusCode).thenReturn(HttpStatus.ok);
+  when(response.contentLength).thenReturn(kTransparentImage.length);
+  when(response.listen(
+    any,
+    onDone: anyNamed('onDone'),
+    onError: anyNamed('onError'),
+    cancelOnError: anyNamed('cancelOnError'),
+  )).thenAnswer((Invocation invocation) {
+    final void Function(Uint8List) onData = invocation.positionalArguments[0];
+    final void Function(Object) onError = invocation.namedArguments[#onError];
+    final void Function() onDone = invocation.namedArguments[#onDone];
+    final bool cancelOnError = invocation.namedArguments[#cancelOnError];
+
+    return Stream<Uint8List>.fromIterable(chunks).listen(
+      onData,
+      onDone: onDone,
+      onError: onError,
+      cancelOnError: cancelOnError,
+    );
+  });
+  return httpClient;
+}
+
+HttpClient respondErrorOnConnection() {
+  final MockHttpClientRequest request = MockHttpClientRequest();
+  final MockHttpClientResponse response = MockHttpClientResponse();
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
+  when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
+  when(response.statusCode).thenReturn(HttpStatus.badGateway);
+  return httpClient;
+}
+
+HttpClient respondErrorOnAny() {
+  const int chunkSize = 8;
+  final MockHttpClientRequest request = MockHttpClientRequest();
+  final MockHttpClientResponse response = MockHttpClientResponse();
+  final MockHttpClient httpClient = MockHttpClient();
+  when(httpClient.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
+  when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
+  when(response.statusCode).thenReturn(HttpStatus.ok);
+  when(response.contentLength).thenReturn(kTransparentImage.length);
+  when(response.listen(
+    any,
+    onDone: anyNamed('onDone'),
+    onError: anyNamed('onError'),
+    cancelOnError: anyNamed('cancelOnError'),
+  )).thenAnswer((Invocation invocation) {
+    final void Function(Uint8List) onData = invocation.positionalArguments[0];
+    final void Function(Object) onError = invocation.namedArguments[#onError];
+    final void Function() onDone = invocation.namedArguments[#onDone];
+    final bool cancelOnError = invocation.namedArguments[#cancelOnError];
+
+    return createRottenChunks(chunkSize).listen(
+      onData,
+      onDone: onDone,
+      onError: onError,
+      cancelOnError: cancelOnError,
+    );
+  });
+  return httpClient;
+}
+
+List<Uint8List> createChunks(int chunkSize) {
+  final List<Uint8List> chunks = <Uint8List>[
+    for (int offset = 0; offset < kTransparentImage.length; offset += chunkSize)
+      Uint8List.fromList(kTransparentImage.skip(offset).take(chunkSize).toList()),
+  ];
+  return chunks;
+}
+
+Stream<Uint8List> createRottenChunks(int chunkSize) async* {
+  yield Uint8List.fromList(kTransparentImage.take(chunkSize).toList());
+  throw 'failed chunk';
+}

packages/flutter/test/widgets/image_headers_test.dart

-// Copyright 2017 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-import 'dart:async';
-import 'dart:io';
-
-import 'package:flutter/widgets.dart';
-import 'package:flutter_test/flutter_test.dart';
-import 'package:mockito/mockito.dart';
-
-import '../painting/image_data.dart';
-
-void main() {
-  final MockHttpClient client = MockHttpClient();
-  final MockHttpClientRequest request = MockHttpClientRequest();
-  final MockHttpClientResponse response = MockHttpClientResponse();
-  final MockHttpHeaders headers = MockHttpHeaders();
-
-  testWidgets('Headers', (WidgetTester tester) async {
-    HttpOverrides.runZoned<Future<void>>(() async {
-      await tester.pumpWidget(Image.network(
-        'https://www.example.com/images/frame.png',
-        headers: const <String, String>{'flutter': 'flutter'},
-      ));
-
-      verify(headers.add('flutter', 'flutter')).called(1);
-
-    }, createHttpClient: (SecurityContext _) {
-      when(client.getUrl(any)).thenAnswer((_) => Future<HttpClientRequest>.value(request));
-      when(request.headers).thenReturn(headers);
-      when(request.close()).thenAnswer((_) => Future<HttpClientResponse>.value(response));
-      when(response.contentLength).thenReturn(kTransparentImage.length);
-      when(response.statusCode).thenReturn(HttpStatus.ok);
-      when(response.listen(any)).thenAnswer((Invocation invocation) {
-        final void Function(List<int>) onData = invocation.positionalArguments[0];
-        final void Function() onDone = invocation.namedArguments[#onDone];
-        final void Function(Object, [ StackTrace ]) onError = invocation.namedArguments[#onError];
-        final bool cancelOnError = invocation.namedArguments[#cancelOnError];
-        return Stream<List<int>>.fromIterable(<List<int>>[kTransparentImage]).listen(onData, onDone: onDone, onError: onError, cancelOnError: cancelOnError);
-      });
-      return client;
-    });
-  }, skip: isBrowser);
-}
-
-class MockHttpClient extends Mock implements HttpClient {}
-
-class MockHttpClientRequest extends Mock implements HttpClientRequest {}
-
-class MockHttpClientResponse extends Mock implements HttpClientResponse {}
-
-class MockHttpHeaders extends Mock implements HttpHeaders {}