use transformer to account for slow producer

testing/scenario_app/android/app/src/androidTest/java/dev/flutter/scenariosui/ScreenshotUtil.java

@@ -37,8 +37,10 @@ private static class Connection {
     }
 
     synchronized void writeFile(String name, byte[] fileContent) throws IOException {
-      final ByteBuffer buffer = ByteBuffer.allocate(name.length() + fileContent.length + 4);
+      final ByteBuffer buffer = ByteBuffer.allocate(name.length() + fileContent.length + 8);
+      // See ScreenshotBlobTransformer#bind in screenshot_transformer.dart
       buffer.putInt(name.length());
+      buffer.putInt(fileContent.length);
       buffer.put(name.getBytes());
       buffer.put(fileContent);
       final byte[] bytes = buffer.array();

testing/scenario_app/bin/android_integration_tests.dart

@@ -3,7 +3,6 @@
 // found in the LICENSE file.
 
 import 'dart:async';
-import 'dart:convert';
 import 'dart:io';
 import 'dart:typed_data';
 
@@ -14,6 +13,7 @@ import 'package:skia_gold_client/skia_gold_client.dart';
 
 import 'utils/logs.dart';
 import 'utils/process_manager_extension.dart';
+import 'utils/screenshot_transformer.dart';
 
 const int tcpPort = 3001;
 
@@ -62,22 +62,21 @@ void main(List<String> args) async {
     stdout.writeln('listening on host ${server.address.address}:${server.port}');
     server.listen((Socket client) {
       stdout.writeln('client connected ${client.remoteAddress.address}:${client.remotePort}');
-      client.listen((Uint8List data) {
-        final int fnameLen = data.buffer.asByteData(0, 4).getInt32(0);
-        final String fileName = utf8.decode(data.buffer.asUint8List(4, fnameLen));
-        final Uint8List fileContent = data.buffer.asUint8List(4 + fnameLen);
+      client.transform(const ScreenshotBlobTransformer()).listen((Screenshot screenshot) {
+        final String fileName = screenshot.filename;
+        final Uint8List fileContent = screenshot.fileContent;
         log('host received ${fileContent.lengthInBytes} bytes for screenshot `$fileName`');
         assert(skiaGoldClient != null, 'expected Skia Gold client');
         late File goldenFile;
         try {
-          goldenFile = File(join(screenshotPath, '$fileName.png'))..writeAsBytesSync(fileContent, flush: true);
+          goldenFile = File(join(screenshotPath, fileName))..writeAsBytesSync(fileContent, flush: true);
         } on FileSystemException catch (err) {
           panic(<String>['failed to create screenshot $fileName: ${err.toString()}']);
         }
         log('wrote ${goldenFile.absolute.path}');
         if (isSkiaGoldClientAvailable) {
           final Future<void> comparison = skiaGoldClient!
-            .addImg('$fileName.png', goldenFile, screenshotSize: fileContent.lengthInBytes)
+            .addImg(fileName, goldenFile, screenshotSize: fileContent.lengthInBytes)
             .catchError((dynamic err) {
               panic(<String>['skia gold comparison failed: ${err.toString()}']);
             });
@@ -201,13 +200,8 @@ void main(List<String> args) async {
       await Future.wait(pendingComparisons);
     });
 
-    await step('Closing logcat...', () async {
-      try {
-        await logcat.flush();
-        await logcat.close();
-      } catch(_) {
-        // ignore.
-      }
+    await step('Flush logcat...', () async {
+      await logcat.flush();
     });
 
     exit(0);

testing/scenario_app/bin/utils/screenshot_transformer.dart

@@ -0,0 +1,66 @@
+// Copyright 2013 The Flutter 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:convert';
+import 'dart:typed_data';
+
+/// A screenshot from the Android emulator.
+class Screenshot {
+  Screenshot(this.filename, this.fileContent);
+
+  /// The name of the screenshot.
+  final String filename;
+
+  /// The binary content of the screenshot.
+  final Uint8List fileContent;
+}
+
+/// Takes the input stream and transforms it into [Screenshot]s.
+class ScreenshotBlobTransformer extends StreamTransformerBase<Uint8List, Screenshot> {
+  const ScreenshotBlobTransformer();
+
+  @override
+  Stream<Screenshot> bind(Stream<Uint8List> stream) async* {
+    final BytesBuilder pending = BytesBuilder();
+
+    await for (final Uint8List blob in stream) {
+      pending.add(blob);
+
+      if (pending.length < 8) {
+        continue;
+      }
+
+      // See ScreenshotUtil#writeFile in ScreenshotUtil.java
+      final Uint8List bytes = pending.toBytes();
+      final ByteData byteData = bytes.buffer.asByteData();
+
+      int off = 0;
+      final int fnameLen = byteData.getInt32(off);
+      off += 4;
+      final int fcontentLen = byteData.getInt32(off);
+      off += 4;
+
+      assert(fnameLen > 0);
+      assert(fcontentLen > 0);
+
+      if (pending.length < off + fnameLen) {
+        continue;
+      }
+
+      final String filename = utf8.decode(bytes.buffer.asUint8List(off, fnameLen));
+      off += fnameLen;
+      if (pending.length < off + fcontentLen) {
+        continue;
+      }
+
+      final Uint8List fileContent = bytes.buffer.asUint8List(off, fcontentLen);
+      off += fcontentLen;
+      pending.clear();
+      pending.add(bytes.buffer.asUint8List(off));
+
+      yield Screenshot('$filename.png', fileContent);
+    }
+  }
+}