Add option in ProxiedDevice to only transfer the delta when deploying. (#97462)

packages/flutter_tools/lib/src/commands/daemon.dart

@@ -27,6 +27,7 @@ import '../emulator.dart';
 import '../features.dart';
 import '../globals.dart' as globals;
 import '../project.dart';
+import '../proxied_devices/file_transfer.dart';
 import '../resident_runner.dart';
 import '../run_cold.dart';
 import '../run_hot.dart';
@@ -1320,6 +1321,8 @@ class EmulatorDomain extends Domain {
 class ProxyDomain extends Domain {
   ProxyDomain(Daemon daemon) : super(daemon, 'proxy') {
     registerHandlerWithBinary('writeTempFile', writeTempFile);
+    registerHandler('calculateFileHashes', calculateFileHashes);
+    registerHandlerWithBinary('updateFile', updateFile);
     registerHandler('connect', connect);
     registerHandler('disconnect', disconnect);
     registerHandlerWithBinary('write', write);
@@ -1336,6 +1339,29 @@ class ProxyDomain extends Domain {
     await file.openWrite().addStream(binary);
   }
 
+  /// Calculate rolling hashes for a file in the local temporary directory.
+  Future<Map<String, dynamic>> calculateFileHashes(Map<String, dynamic> args) async {
+    final String path = _getStringArg(args, 'path', required: true);
+    final File file = tempDirectory.childFile(path);
+    if (!await file.exists()) {
+      return null;
+    }
+    final BlockHashes result = await FileTransfer().calculateBlockHashesOfFile(file);
+    return result.toJson();
+  }
+
+  Future<bool> updateFile(Map<String, dynamic> args, Stream<List<int>> binary) async {
+    final String path = _getStringArg(args, 'path', required: true);
+    final File file = tempDirectory.childFile(path);
+    if (!await file.exists()) {
+      return null;
+    }
+    final List<Map<String, dynamic>> deltaJson = (args['delta'] as List<dynamic>).cast<Map<String, dynamic>>();
+    final List<FileDeltaBlock> delta = FileDeltaBlock.fromJsonList(deltaJson);
+    final bool result = await FileTransfer().rebuildFile(file, delta, binary);
+    return result;
+  }
+
   /// Opens a connection to a local port, and returns the connection id.
   Future<String> connect(Map<String, dynamic> args) async {
     final int targetPort = _getIntArg(args, 'port', required: true);
@@ -1404,12 +1430,13 @@ class ProxyDomain extends Domain {
     for (final Socket connection in _forwardedConnections.values) {
       connection.destroy();
     }
-    await _tempDirectory?.delete(recursive: true);
+    // We deliberately not clean up the tempDirectory here. The application package files that
+    // are transferred into this directory through ProxiedDevices are left in the directory
+    // to be reused on any subsequent runs.
   }
 
-
   Directory _tempDirectory;
-  Directory get tempDirectory => _tempDirectory ??= globals.fs.systemTempDirectory.createTempSync('flutter_tool_daemon.');
+  Directory get tempDirectory => _tempDirectory ??= globals.fs.systemTempDirectory.childDirectory('flutter_tool_daemon')..createSync();
 }
 
 /// A [Logger] which sends log messages to a listening daemon client.

packages/flutter_tools/lib/src/proxied_devices/devices.dart

@@ -16,6 +16,7 @@ import '../daemon.dart';
 import '../device.dart';
 import '../device_port_forwarder.dart';
 import '../project.dart';
+import 'file_transfer.dart';
 
 bool _isNullable<T>() => null is T;
 
@@ -29,16 +30,23 @@ T _cast<T>(Object? object) {
 
 /// A [DeviceDiscovery] that will connect to a flutter daemon and connects to
 /// the devices remotely.
+///
+/// If [deltaFileTransfer] is true, the proxy will use an rsync-like algorithm that
+/// only transfers the changed part of the application package for deployment.
 class ProxiedDevices extends DeviceDiscovery {
   ProxiedDevices(this.connection, {
+    bool deltaFileTransfer = true,
     required Logger logger,
-  }) : _logger = logger;
+  }) : _deltaFileTransfer = deltaFileTransfer,
+       _logger = logger;
 
   /// [DaemonConnection] used to communicate with the daemon.
   final DaemonConnection connection;
 
   final Logger _logger;
 
+  final bool _deltaFileTransfer;
+
   @override
   bool get supportsPlatform => true;
 
@@ -70,6 +78,7 @@ class ProxiedDevices extends DeviceDiscovery {
     final Map<String, Object?> capabilities = _cast<Map<String, Object?>>(device['capabilities']);
     return ProxiedDevice(
       connection, _cast<String>(device['id']),
+      deltaFileTransfer: _deltaFileTransfer,
       category: Category.fromString(_cast<String>(device['category'])),
       platformType: PlatformType.fromString(_cast<String>(device['platformType'])),
       targetPlatform: getTargetPlatformForName(_cast<String>(device['platform'])),
@@ -92,8 +101,12 @@ class ProxiedDevices extends DeviceDiscovery {
 /// A [Device] that acts as a proxy to remotely connected device.
 ///
 /// The communication happens via a flutter daemon.
+///
+/// If [deltaFileTransfer] is true, the proxy will use an rsync-like algorithm that
+/// only transfers the changed part of the application package for deployment.
 class ProxiedDevice extends Device {
   ProxiedDevice(this.connection, String id, {
+    bool deltaFileTransfer = true,
     required Category? category,
     required PlatformType? platformType,
     required TargetPlatform targetPlatform,
@@ -109,7 +122,8 @@ class ProxiedDevice extends Device {
     required this.supportsFastStart,
     required bool supportsHardwareRendering,
     required Logger logger,
-  }): _isLocalEmulator = isLocalEmulator,
+  }): _deltaFileTransfer = deltaFileTransfer,
+      _isLocalEmulator = isLocalEmulator,
       _emulatorId = emulatorId,
       _sdkNameAndVersion = sdkNameAndVersion,
       _supportsHardwareRendering = supportsHardwareRendering,
@@ -125,6 +139,8 @@ class ProxiedDevice extends Device {
 
   final Logger _logger;
 
+  final bool _deltaFileTransfer;
+
   @override
   final String name;
 
@@ -288,9 +304,37 @@ class ProxiedDevice extends Device {
     final String fileName = binary.basename;
     final Completer<String> idCompleter = Completer<String>();
     _applicationPackageMap[path] = idCompleter.future;
-    await connection.sendRequest('proxy.writeTempFile', <String, Object>{
+
+    final Map<String, Object> args = <String, Object>{'path': fileName};
+
+    Map<String, Object?>? rollingHashResultJson;
+    if (_deltaFileTransfer) {
+     rollingHashResultJson = _cast<Map<String, Object?>?>(await connection.sendRequest('proxy.calculateFileHashes', args));
+    }
+
+    if (rollingHashResultJson == null) {
+      // Either file not found on the remote end, or deltaFileTransfer is set to false, transfer the file directly.
+      if (_deltaFileTransfer) {
+        _logger.printTrace('Delta file transfer is enabled but file is not found on the remote end, do a full transfer.');
+      }
+
+      await connection.sendRequest('proxy.writeTempFile', args, await binary.readAsBytes());
+    } else {
+      final BlockHashes rollingHashResult = BlockHashes.fromJson(rollingHashResultJson);
+      final List<FileDeltaBlock> delta = await FileTransfer().computeDelta(binary, rollingHashResult);
+
+      // Delta is empty if the file does not need to be updated
+      if (delta.isNotEmpty) {
+        final List<Map<String, Object>> deltaJson = delta.map((FileDeltaBlock block) => block.toJson()).toList();
+        final Uint8List buffer = await FileTransfer().binaryForRebuilding(binary, delta);
+
+        await connection.sendRequest('proxy.updateFile', <String, Object>{
           'path': fileName,
-    }, await binary.readAsBytes());
+          'delta': deltaJson,
+        }, buffer);
+      }
+    }
+
     final String id = _cast<String>(await connection.sendRequest('device.uploadApplicationPackage', <String, Object>{
       'targetPlatform': getNameForTargetPlatform(_targetPlatform),
       'applicationBinary': fileName,

packages/flutter_tools/test/general.shard/proxied_devices/file_transfer_test.dart

+// Copyright 2014 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:typed_data';
+
+import 'package:file/memory.dart';
+import 'package:flutter_tools/src/base/file_system.dart';
+import 'package:flutter_tools/src/convert.dart';
+import 'package:flutter_tools/src/proxied_devices/file_transfer.dart';
+
+import '../../src/common.dart';
+
+void main() {
+  // late BufferLogger bufferLogger;
+  // late FakeDaemonStreams daemonStreams;
+  // late DaemonConnection daemonConnection;
+  // setUp(() {
+  //   bufferLogger = BufferLogger.test();
+  //   daemonStreams = FakeDaemonStreams();
+  //   daemonConnection = DaemonConnection(
+  //     daemonStreams: daemonStreams,
+  //     logger: bufferLogger,
+  //   );
+  // });
+
+  group('convertToChunks', () {
+    test('works correctly', () async {
+      final StreamController<Uint8List> controller = StreamController<Uint8List>();
+      final Stream<Uint8List> chunked = convertToChunks(controller.stream, 4);
+      final Future<List<Uint8List>> chunkedListFuture = chunked.toList();
+
+      // Full chunk.
+      controller.add(Uint8List.fromList(<int>[1, 2, 3, 4]));
+      // Multiple of full chunks, on chunk bounraries.
+      controller.add(Uint8List.fromList(<int>[5, 6, 7, 8, 9, 10, 11, 12]));
+      // Larger than one chunk, starts on chunk boundary, ends not on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[13, 14, 15, 16, 17, 18]));
+      // Larger than one chunk, starts not on chunk boundary, ends not on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[19, 20, 21, 22, 23]));
+      // Larger than one chunk, starts not on chunk boundary, ends on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[24, 25, 26, 27, 28]));
+      // Smaller than one chunk, starts on chunk boundary, ends not on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[29, 30]));
+      // Smaller than one chunk, starts not on chunk boundary, ends not on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[31, 32, 33]));
+      // Full chunk, not on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[34, 35, 36, 37]));
+      // Smaller than one chunk, starts not on chunk boundary, ends on chunk boundary.
+      controller.add(Uint8List.fromList(<int>[38, 39, 40]));
+      // Empty chunk.
+      controller.add(Uint8List.fromList(<int>[]));
+      // Extra chunk.
+      controller.add(Uint8List.fromList(<int>[41, 42]));
+
+      await controller.close();
+
+      final List<Uint8List> chunkedList = await chunkedListFuture;
+
+      expect(chunkedList, hasLength(11));
+      expect(chunkedList[0], <int>[1, 2, 3, 4]);
+      expect(chunkedList[1], <int>[5, 6, 7, 8]);
+      expect(chunkedList[2], <int>[9, 10, 11, 12]);
+      expect(chunkedList[3], <int>[13, 14, 15, 16]);
+      expect(chunkedList[4], <int>[17, 18, 19, 20]);
+      expect(chunkedList[5], <int>[21, 22, 23, 24]);
+      expect(chunkedList[6], <int>[25, 26, 27, 28]);
+      expect(chunkedList[7], <int>[29, 30, 31, 32]);
+      expect(chunkedList[8], <int>[33, 34, 35, 36]);
+      expect(chunkedList[9], <int>[37, 38, 39, 40]);
+      expect(chunkedList[10], <int>[41, 42]);
+    });
+  });
+
+  group('adler32Hash', () {
+    test('works correctly', () {
+      final int hash = adler32Hash(utf8.encode('abcdefg'));
+      expect(hash, 0x0adb02bd);
+    });
+  });
+
+  group('RollingAdler32', () {
+    test('works correctly without rolling', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('abcdefg').forEach(adler32.push);
+      expect(adler32.hash, adler32Hash(utf8.encode('abcdefg')));
+    });
+
+    test('works correctly after rolling once', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('12abcdefg').forEach(adler32.push);
+      expect(adler32.hash, adler32Hash(utf8.encode('abcdefg')));
+    });
+
+    test('works correctly after rolling multiple cycles', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('1234567890123456789abcdefg').forEach(adler32.push);
+      expect(adler32.hash, adler32Hash(utf8.encode('abcdefg')));
+    });
+
+    test('works correctly after reset', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('1234567890123456789abcdefg').forEach(adler32.push);
+      adler32.reset();
+      utf8.encode('abcdefg').forEach(adler32.push);
+      expect(adler32.hash, adler32Hash(utf8.encode('abcdefg')));
+    });
+
+    test('currentBlock returns the correct entry when read less than one block', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('abcd').forEach(adler32.push);
+      expect(adler32.currentBlock(), utf8.encode('abcd'));
+    });
+
+    test('currentBlock returns the correct entry when read exactly one block', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('abcdefg').forEach(adler32.push);
+      expect(adler32.currentBlock(), utf8.encode('abcdefg'));
+    });
+
+    test('currentBlock returns the correct entry when read more than one block', () {
+      final RollingAdler32 adler32 = RollingAdler32(7);
+      utf8.encode('123456789abcdefg').forEach(adler32.push);
+      expect(adler32.currentBlock(), utf8.encode('abcdefg'));
+    });
+  });
+
+  group('FileTransfer', () {
+    const String content1 = 'a...b...c...d...e.';
+    const String content2 = 'b...c...d...a...f...g...b...h..';
+    const List<FileDeltaBlock> expectedDelta = <FileDeltaBlock>[
+      FileDeltaBlock.fromDestination(start: 4, size: 12),
+      FileDeltaBlock.fromDestination(start: 0, size: 4),
+      FileDeltaBlock.fromSource(start: 16, size: 8),
+      FileDeltaBlock.fromDestination(start: 4, size: 4),
+      FileDeltaBlock.fromSource(start: 28, size: 3),
+    ];
+    const String expectedBinaryForRebuilding = 'f...g...h..';
+    late MemoryFileSystem fileSystem;
+    setUp(() {
+      fileSystem = MemoryFileSystem();
+    });
+
+    test('calculateBlockHashesOfFile works normally', () async {
+      final File file = fileSystem.file('test')..writeAsStringSync(content1);
+
+      final BlockHashes hashes = await FileTransfer().calculateBlockHashesOfFile(file, blockSize: 4);
+      expect(hashes.blockSize, 4);
+      expect(hashes.totalSize, content1.length);
+      expect(hashes.adler32, hasLength(5));
+      expect(hashes.adler32, <int>[
+        0x029c00ec,
+        0x02a000ed,
+        0x02a400ee,
+        0x02a800ef,
+        0x00fa0094,
+      ]);
+      expect(hashes.md5, hasLength(5));
+      expect(hashes.md5, <String>[
+        'zB0S8R/fGt05GcI5v8AjIQ==',
+        'uZCZ4i/LUGFYAD+K1ZD0Wg==',
+        '6kbZGS8T1NJl/naWODQcNw==',
+        'kKh/aA2XAhR/r0HdZa3Bxg==',
+        '34eF7Bs/OhfoJ5+sAw0zyw==',
+      ]);
+      expect(hashes.fileMd5, 'VT/gkSEdctzUEUJCxclxuQ==');
+    });
+
+    test('computeDelta returns empty list if file is identical', () async {
+      final File file1 = fileSystem.file('file1')..writeAsStringSync(content1);
+      final File file2 = fileSystem.file('file1')..writeAsStringSync(content1);
+
+      final BlockHashes hashes = await FileTransfer().calculateBlockHashesOfFile(file1, blockSize: 4);
+      final List<FileDeltaBlock> delta = await FileTransfer().computeDelta(file2, hashes);
+
+      expect(delta, isEmpty);
+    });
+
+    test('computeDelta returns the correct delta', () async {
+      final File file1 = fileSystem.file('file1')..writeAsStringSync(content1);
+      final File file2 = fileSystem.file('file2')..writeAsStringSync(content2);
+
+      final BlockHashes hashes = await FileTransfer().calculateBlockHashesOfFile(file1, blockSize: 4);
+      final List<FileDeltaBlock> delta = await FileTransfer().computeDelta(file2, hashes);
+
+      expect(delta, expectedDelta);
+    });
+
+    test('binaryForRebuilding returns the correct binary', () async {
+      final File file = fileSystem.file('file')..writeAsStringSync(content2);
+      final List<int> binaryForRebuilding = await FileTransfer().binaryForRebuilding(file, expectedDelta);
+      expect(binaryForRebuilding, utf8.encode(expectedBinaryForRebuilding));
+    });
+
+    test('rebuildFile can rebuild the correct file', () async {
+      final File file = fileSystem.file('file')..writeAsStringSync(content1);
+      await FileTransfer().rebuildFile(file, expectedDelta, Stream<List<int>>.fromIterable(<List<int>>[utf8.encode(expectedBinaryForRebuilding)]));
+      expect(file.readAsStringSync(), content2);
+    });
+  });
+}