Allow daemon to start DDS remotely when proxied devices are used. (#124061)

Allow daemon to start DDS remotely when proxied devices are used.

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

@@ -839,6 +839,9 @@ class DeviceDomain extends Domain {
     registerHandler('startApp', startApp);
     registerHandler('stopApp', stopApp);
     registerHandler('takeScreenshot', takeScreenshot);
+    registerHandler('startDartDevelopmentService', startDartDevelopmentService);
+    registerHandler('shutdownDartDevelopmentService', shutdownDartDevelopmentService);
+    registerHandler('setExternalDevToolsUriForDartDevelopmentService', setExternalDevToolsUriForDartDevelopmentService);
 
     // Use the device manager discovery so that client provided device types
     // are usable via the daemon protocol.
@@ -1059,6 +1062,50 @@ class DeviceDomain extends Domain {
     }
   }
 
+  /// Starts DDS for the device.
+  Future<String?> startDartDevelopmentService(Map<String, Object?> args) async {
+    final String? deviceId = _getStringArg(args, 'deviceId', required: true);
+    final bool? disableServiceAuthCodes = _getBoolArg(args, 'disableServiceAuthCodes');
+    final String vmServiceUriStr = _getStringArg(args, 'vmServiceUri', required: true)!;
+
+    final Device? device = await daemon.deviceDomain._getDevice(deviceId);
+    if (device == null) {
+      throw DaemonException("device '$deviceId' not found");
+    }
+
+    await device.dds.startDartDevelopmentService(
+      Uri.parse(vmServiceUriStr),
+      logger: globals.logger,
+      disableServiceAuthCodes: disableServiceAuthCodes,
+    );
+    unawaited(device.dds.done.whenComplete(() => sendEvent('device.dds.done.$deviceId')));
+    return device.dds.uri?.toString();
+  }
+
+  /// Starts DDS for the device.
+  Future<void> shutdownDartDevelopmentService(Map<String, Object?> args) async {
+    final String? deviceId = _getStringArg(args, 'deviceId', required: true);
+
+    final Device? device = await daemon.deviceDomain._getDevice(deviceId);
+    if (device == null) {
+      throw DaemonException("device '$deviceId' not found");
+    }
+
+    await device.dds.shutdown();
+  }
+
+  Future<void> setExternalDevToolsUriForDartDevelopmentService(Map<String, Object?> args) async {
+    final String? deviceId = _getStringArg(args, 'deviceId', required: true);
+    final String uri = _getStringArg(args, 'uri', required: true)!;
+
+    final Device? device = await daemon.deviceDomain._getDevice(deviceId);
+    if (device == null) {
+      throw DaemonException("device '$deviceId' not found");
+    }
+
+    device.dds.setExternalDevToolsUri(Uri.parse(uri));
+  }
+
   @override
   Future<void> dispose() {
     for (final PollingDeviceDiscovery discoverer in _discoverers) {

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

@@ -8,6 +8,7 @@ import 'dart:typed_data';
 import 'package:meta/meta.dart';
 
 import '../application_package.dart';
+import '../base/dds.dart';
 import '../base/file_system.dart';
 import '../base/io.dart';
 import '../base/logger.dart';
@@ -37,8 +38,10 @@ T _cast<T>(Object? object) {
 class ProxiedDevices extends DeviceDiscovery {
   ProxiedDevices(this.connection, {
     bool deltaFileTransfer = true,
+    bool enableDdsProxy = false,
     required Logger logger,
   }) : _deltaFileTransfer = deltaFileTransfer,
+       _enableDdsProxy = enableDdsProxy,
        _logger = logger;
 
   /// [DaemonConnection] used to communicate with the daemon.
@@ -48,6 +51,8 @@ class ProxiedDevices extends DeviceDiscovery {
 
   final bool _deltaFileTransfer;
 
+  final bool _enableDdsProxy;
+
   @override
   bool get supportsPlatform => true;
 
@@ -91,6 +96,7 @@ class ProxiedDevices extends DeviceDiscovery {
     return ProxiedDevice(
       connection, _cast<String>(device['id']),
       deltaFileTransfer: _deltaFileTransfer,
+      enableDdsProxy: _enableDdsProxy,
       category: Category.fromString(_cast<String>(device['category'])),
       platformType: PlatformType.fromString(_cast<String>(device['platformType'])),
       targetPlatform: getTargetPlatformForName(_cast<String>(device['platform'])),
@@ -116,9 +122,13 @@ class ProxiedDevices extends DeviceDiscovery {
 ///
 /// If [deltaFileTransfer] is true, the proxy will use an rsync-like algorithm that
 /// only transfers the changed part of the application package for deployment.
+///
+/// If [enableDdsProxy] is true, DDS will be started on the daemon instead of
+/// starting locally.
 class ProxiedDevice extends Device {
   ProxiedDevice(this.connection, String id, {
     bool deltaFileTransfer = true,
+    bool enableDdsProxy = false,
     required Category? category,
     required PlatformType? platformType,
     required TargetPlatform targetPlatform,
@@ -135,6 +145,7 @@ class ProxiedDevice extends Device {
     required bool supportsHardwareRendering,
     required Logger logger,
   }): _deltaFileTransfer = deltaFileTransfer,
+      _enableDdsProxy = enableDdsProxy,
       _isLocalEmulator = isLocalEmulator,
       _emulatorId = emulatorId,
       _sdkNameAndVersion = sdkNameAndVersion,
@@ -153,6 +164,8 @@ class ProxiedDevice extends Device {
 
   final bool _deltaFileTransfer;
 
+  final bool _enableDdsProxy;
+
   @override
   final String name;
 
@@ -229,6 +242,16 @@ class ProxiedDevice extends Device {
   @override
   DevicePortForwarder get portForwarder => _portForwarder ??= ProxiedPortForwarder(connection, deviceId: id, logger: _logger);
 
+  ProxiedDartDevelopmentService? _proxiedDds;
+  @override
+  DartDevelopmentService get dds {
+    if (!_enableDdsProxy) {
+      return super.dds;
+    }
+    return _proxiedDds ??= ProxiedDartDevelopmentService(connection, id,
+        logger: _logger, proxiedPortForwarder: proxiedPortForwarder);
+  }
+
   @override
   void clearLogs() => throw UnimplementedError();
 
@@ -310,10 +333,14 @@ class ProxiedDevice extends Device {
   }
 
   @override
-  Future<void> dispose() async {}
+  Future<void> dispose() async {
+    await proxiedPortForwarder.dispose();
+  }
 
-  final Map<String, Future<String>> _applicationPackageMap = <String, Future<String>>{};
-  Future<String> applicationPackageId(PrebuiltApplicationPackage package) async {
+  final Map<String, Future<String>> _applicationPackageMap =
+      <String, Future<String>>{};
+  Future<String> applicationPackageId(
+      PrebuiltApplicationPackage package) async {
     final File binary = package.applicationPackage as File;
     final String path = binary.absolute.path;
     if (_applicationPackageMap.containsKey(path)) {
@@ -449,7 +476,7 @@ class _ProxiedForwardedPort extends ForwardedPort {
   }
 }
 
-typedef CreateSocketServer = Future<ServerSocket> Function(Logger logger, int? hostPort);
+typedef CreateSocketServer = Future<ServerSocket> Function(Logger logger, int? hostPort, bool? ipv6);
 
 /// A [DevicePortForwarder] for a proxied device.
 ///
@@ -484,7 +511,7 @@ class ProxiedPortForwarder extends DevicePortForwarder {
   final List<Socket> _connectedSockets = <Socket>[];
 
   @override
-  Future<int> forward(int devicePort, { int? hostPort }) async {
+  Future<int> forward(int devicePort, {int? hostPort, bool? ipv6}) async {
     int? remoteDevicePort;
     final String? deviceId = _deviceId;
 
@@ -500,7 +527,7 @@ class ProxiedPortForwarder extends DevicePortForwarder {
       devicePort = result['hostPort']! as int;
     }
 
-    final ServerSocket serverSocket = await _startProxyServer(devicePort, hostPort);
+    final ServerSocket serverSocket = await _startProxyServer(devicePort, hostPort, ipv6);
 
     _hostPortToForwardedPorts[serverSocket.port] = _ProxiedForwardedPort(
       connection,
@@ -514,8 +541,8 @@ class ProxiedPortForwarder extends DevicePortForwarder {
     return serverSocket.port;
   }
 
-  Future<ServerSocket> _startProxyServer(int devicePort, int? hostPort) async {
-    final ServerSocket serverSocket = await _createSocketServer(_logger, hostPort);
+  Future<ServerSocket> _startProxyServer(int devicePort, int? hostPort, bool? ipv6) async {
+    final ServerSocket serverSocket = await _createSocketServer(_logger, hostPort, ipv6);
 
     serverSocket.listen((Socket socket) async {
       final String id = _cast<String>(await connection.sendRequest('proxy.connect', <String, Object>{
@@ -595,20 +622,171 @@ class ProxiedPortForwarder extends DevicePortForwarder {
       await forwardedPort.unforward();
     }
 
-    for (final Socket socket in _connectedSockets) {
-      await socket.close();
-    }
+    await Future.wait(<Future<void>>[
+      for (final Socket socket in _connectedSockets)
+        socket.close(),
+    ]);
+  }
+
+  /// Returns the original remote port given the local port.
+  ///
+  /// If this is not a port that is handled by this port forwarder, return null.
+  int? originalRemotePort(int localForwardedPort) {
+    return _hostPortToForwardedPorts[localForwardedPort]?.devicePort;
   }
 }
 
-Future<ServerSocket> _defaultCreateServerSocket(Logger logger, int? hostPort) async {
-  try {
-    return await ServerSocket.bind(InternetAddress.loopbackIPv4, hostPort ?? 0);
-  } on SocketException {
-    logger.printTrace('Bind on $hostPort failed with IPv4, retrying on IPv6');
+Future<ServerSocket> _defaultCreateServerSocket(Logger logger, int? hostPort, bool? ipv6) async {
+  if (ipv6 == null || ipv6 == false) {
+    try {
+      return await ServerSocket.bind(InternetAddress.loopbackIPv4, hostPort ?? 0);
+    } on SocketException {
+      logger.printTrace('Bind on $hostPort failed with IPv4, retrying on IPv6');
+    }
   }
 
   // If binding on ipv4 failed, try binding on ipv6.
   // Omit try catch here, let the failure fallthrough.
   return ServerSocket.bind(InternetAddress.loopbackIPv6, hostPort ?? 0);
 }
+
+/// A class that starts the [DartDevelopmentService] on the daemon.
+///
+/// There are a lot of communications between DDS and the VM service on the
+/// device. When using proxied device, starting DDS remotely helps reduces the
+/// amount of data transferred with the remote daemon, hence improving latency.
+class ProxiedDartDevelopmentService implements DartDevelopmentService {
+  ProxiedDartDevelopmentService(
+    this.connection,
+    this.deviceId, {
+    required Logger logger,
+    required ProxiedPortForwarder proxiedPortForwarder,
+    @visibleForTesting DartDevelopmentService? localDds,
+  })  : _logger = logger,
+        _proxiedPortForwarder = proxiedPortForwarder,
+        _localDds = localDds ?? DartDevelopmentService();
+
+  final String deviceId;
+
+  final Logger _logger;
+
+  /// [DaemonConnection] used to communicate with the daemon.
+  final DaemonConnection connection;
+
+  final ProxiedPortForwarder _proxiedPortForwarder;
+
+  Uri? _localUri;
+
+  @override
+  Uri? get uri => _ddsStartedLocally ? _localDds.uri : _localUri;
+
+  @override
+  Future<void> get done => _completer.future;
+  final Completer<void> _completer = Completer<void>();
+
+  final DartDevelopmentService _localDds;
+
+  bool _ddsStartedLocally = false;
+
+  @override
+  Future<void> startDartDevelopmentService(
+    Uri vmServiceUri, {
+    required Logger logger,
+    int? hostPort,
+    bool? ipv6,
+    bool? disableServiceAuthCodes,
+    bool cacheStartupProfile = false,
+  }) async {
+    // Locate the original VM service port on the remote daemon.
+    final int? remoteVMServicePort = _proxiedPortForwarder.originalRemotePort(vmServiceUri.port);
+
+    if (remoteVMServicePort == null) {
+      _logger.printTrace('VM service port is not a forwarded port. Start DDS locally.');
+      _ddsStartedLocally = true;
+      await _localDds.startDartDevelopmentService(
+        vmServiceUri,
+        logger: logger,
+        hostPort: hostPort,
+        ipv6: ipv6,
+        disableServiceAuthCodes: disableServiceAuthCodes,
+        cacheStartupProfile: cacheStartupProfile,
+      );
+      unawaited(_localDds.done.then(_completer.complete));
+      return;
+    }
+
+    final Uri remoteVMServiceUri = vmServiceUri.replace(port: remoteVMServicePort);
+
+    String? remoteUriStr;
+    const String method = 'device.startDartDevelopmentService';
+    try {
+      // Proxies the `done` future.
+      unawaited(connection
+          .listenToEvent('device.dds.done.$deviceId')
+          .first
+          .then(
+            (DaemonEventData event) => _completer.complete(),
+            onError: (_) {
+              // Ignore if we did not receive any event from the server.
+            },
+          ));
+      remoteUriStr = _cast<String?>(await connection.sendRequest(method, <String, Object?>{
+        'deviceId': deviceId,
+        'vmServiceUri': remoteVMServiceUri.toString(),
+        'disableServiceAuthCodes': disableServiceAuthCodes,
+      }));
+    } on String catch (e) {
+      if (!e.contains(method)) {
+        rethrow;
+      }
+      // Remote daemon does not support the command, ignore.
+      // We will try to start DDS locally below.
+    }
+
+    if (remoteUriStr == null) {
+      _logger.printTrace('Remote daemon cannot start DDS. Start a local DDS instead.');
+      _ddsStartedLocally = true;
+      await _localDds.startDartDevelopmentService(
+        vmServiceUri,
+        logger: logger,
+        hostPort: hostPort,
+        ipv6: ipv6,
+        disableServiceAuthCodes: disableServiceAuthCodes,
+        cacheStartupProfile: cacheStartupProfile,
+      );
+      unawaited(_localDds.done.then(_completer.complete));
+      return;
+    }
+
+    _logger.printTrace('Remote DDS started on $remoteUriStr.');
+
+    // Forward the port.
+    final Uri remoteUri = Uri.parse(remoteUriStr);
+    final int localPort = await _proxiedPortForwarder.forward(
+      remoteUri.port,
+      hostPort: hostPort,
+      ipv6: ipv6,
+    );
+
+    _localUri = remoteUri.replace(port: localPort);
+    _logger.printTrace('Local port forwarded DDS on $_localUri.');
+  }
+
+  @override
+  Future<void> shutdown() async {
+    if (_ddsStartedLocally) {
+      await _localDds.shutdown();
+      _ddsStartedLocally = false;
+    } else {
+      await connection.sendRequest('device.shutdownDartDevelopmentService');
+    }
+  }
+
+  @override
+  void setExternalDevToolsUri(Uri uri) {
+    connection.sendRequest('device.setExternalDevToolsUriForDartDevelopmentService', <String, Object?>{
+      'deviceId': deviceId,
+      'uri': uri.toString(),
+    });
+  }
+}

packages/flutter_tools/test/commands.shard/hermetic/daemon_test.dart

@@ -11,6 +11,7 @@ import 'package:file/src/interface/file.dart';
 import 'package:flutter_tools/src/android/android_device.dart';
 import 'package:flutter_tools/src/android/android_workflow.dart';
 import 'package:flutter_tools/src/application_package.dart';
+import 'package:flutter_tools/src/base/dds.dart';
 import 'package:flutter_tools/src/base/logger.dart';
 import 'package:flutter_tools/src/base/utils.dart';
 import 'package:flutter_tools/src/build_info.dart';
@@ -467,6 +468,64 @@ void main() {
       });
     });
 
+    testUsingContext('device.startDartDevelopmentService and .shutdownDartDevelopmentService starts and stops DDS', () async {
+      daemon = Daemon(
+        daemonConnection,
+        notifyingLogger: notifyingLogger,
+      );
+      final FakePollingDeviceDiscovery discoverer = FakePollingDeviceDiscovery();
+      daemon.deviceDomain.addDeviceDiscoverer(discoverer);
+      final FakeAndroidDevice device = FakeAndroidDevice();
+      discoverer.addDevice(device);
+
+      final Completer<void> ddsDoneCompleter = Completer<void>();
+      device.dds.done = ddsDoneCompleter.future;
+      final Uri fakeDdsUri = Uri.parse('http://fake_dds_uri');
+      device.dds.uri = fakeDdsUri;
+
+      // Try starting DDS.
+      expect(device.dds.startCalled, false);
+      daemonStreams.inputs.add(DaemonMessage(<String, Object?>{
+        'id': 0,
+        'method': 'device.startDartDevelopmentService',
+        'params': <String, Object?>{
+          'deviceId': 'device',
+          'disableServiceAuthCodes': false,
+          'vmServiceUri': 'http://fake_uri/auth_code',
+        },
+      }));
+      final Stream<DaemonMessage> broadcastOutput = daemonStreams.outputs.stream.asBroadcastStream();
+      final DaemonMessage startResponse = await broadcastOutput.firstWhere(_notEvent);
+      expect(startResponse.data['id'], 0);
+      expect(startResponse.data['error'], isNull);
+      final String? ddsUri = startResponse.data['result'] as String?;
+      expect(ddsUri, fakeDdsUri.toString());
+      expect(device.dds.startCalled, true);
+      expect(device.dds.startDisableServiceAuthCodes, false);
+      expect(device.dds.startVMServiceUri, Uri.parse('http://fake_uri/auth_code'));
+
+      // dds.done event should be sent to the client.
+      ddsDoneCompleter.complete();
+      final DaemonMessage startEvent = await broadcastOutput.firstWhere(
+        (DaemonMessage message) => message.data['event'] != null && message.data['event'] == 'device.dds.done.device',
+      );
+      expect(startEvent, isNotNull);
+
+      // Try stopping DDS.
+      expect(device.dds.shutdownCalled, false);
+      daemonStreams.inputs.add(DaemonMessage(<String, Object?>{
+        'id': 1,
+        'method': 'device.shutdownDartDevelopmentService',
+        'params': <String, Object?>{
+          'deviceId': 'device',
+        },
+      }));
+      final DaemonMessage stopResponse = await broadcastOutput.firstWhere(_notEvent);
+      expect(stopResponse.data['id'], 1);
+      expect(stopResponse.data['error'], isNull);
+      expect(device.dds.shutdownCalled, true);
+    });
+
     testUsingContext('emulator.launch without an emulatorId should report an error', () async {
       daemon = Daemon(
         daemonConnection,
@@ -877,6 +936,9 @@ class FakeAndroidDevice extends Fake implements AndroidDevice {
   @override
   bool get supportsStartPaused => true;
 
+  @override
+  final FakeDartDevelopmentService dds = FakeDartDevelopmentService();
+
   BuildMode? supportsRuntimeModeCalledBuildMode;
   @override
   Future<bool> supportsRuntimeMode(BuildMode buildMode) async {
@@ -920,6 +982,39 @@ class FakeAndroidDevice extends Fake implements AndroidDevice {
   }
 }
 
+class FakeDartDevelopmentService extends Fake implements DartDevelopmentService {
+  bool startCalled = false;
+  late Uri startVMServiceUri;
+  bool? startDisableServiceAuthCodes;
+
+  bool shutdownCalled = false;
+
+  @override
+  late Future<void> done;
+
+  @override
+  Uri? uri;
+
+  @override
+  Future<void> startDartDevelopmentService(
+    Uri vmServiceUri, {
+    required Logger logger,
+    int? hostPort,
+    bool? ipv6,
+    bool? disableServiceAuthCodes,
+    bool cacheStartupProfile = false,
+  }) async {
+    startCalled = true;
+    startVMServiceUri = vmServiceUri;
+    startDisableServiceAuthCodes = disableServiceAuthCodes;
+  }
+
+  @override
+  Future<void> shutdown() async {
+    shutdownCalled = true;
+  }
+}
+
 class FakeDeviceLogReader implements DeviceLogReader {
   final StreamController<String> logLinesController = StreamController<String>();
   bool disposeCalled = false;

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

@@ -6,6 +6,7 @@ import 'dart:async';
 import 'dart:io';
 import 'dart:typed_data';
 
+import 'package:flutter_tools/src/base/dds.dart';
 import 'package:flutter_tools/src/base/logger.dart';
 import 'package:flutter_tools/src/daemon.dart';
 import 'package:flutter_tools/src/device.dart';
@@ -47,7 +48,7 @@ void main() {
       final ProxiedPortForwarder portForwarder = ProxiedPortForwarder(
         clientDaemonConnection,
         logger: bufferLogger,
-        createSocketServer: (Logger logger, int? hostPort) async =>
+        createSocketServer: (Logger logger, int? hostPort, bool? ipv6) async =>
             fakeServerSocket,
       );
       final int result = await portForwarder.forward(100);
@@ -99,7 +100,7 @@ void main() {
           },
         ),
         logger: bufferLogger,
-        createSocketServer: (Logger logger, int? hostPort) async =>
+        createSocketServer: (Logger logger, int? hostPort, bool? ipv6) async =>
             fakeServerSocket,
       );
       final int result = await portForwarder.forward(100);
@@ -118,7 +119,7 @@ void main() {
         clientDaemonConnection,
         deviceId: 'device_id',
         logger: bufferLogger,
-        createSocketServer: (Logger logger, int? hostPort) async =>
+        createSocketServer: (Logger logger, int? hostPort, bool? ipv6) async =>
             fakeServerSocket,
       );
 
@@ -172,7 +173,7 @@ void main() {
           clientDaemonConnection,
           deviceId: 'device_id',
           logger: bufferLogger,
-          createSocketServer: (Logger logger, int? hostPort) async =>
+          createSocketServer: (Logger logger, int? hostPort, bool? ipv6) async =>
               fakeServerSocket,
         );
 
@@ -225,6 +226,53 @@ void main() {
         await pumpEventQueue();
       });
     });
+
+    testWithoutContext('disposes multiple sockets correctly', () async {
+      final FakeServerSocket fakeServerSocket = FakeServerSocket(200);
+      final ProxiedPortForwarder portForwarder = ProxiedPortForwarder(
+        clientDaemonConnection,
+        logger: bufferLogger,
+        createSocketServer: (Logger logger, int? hostPort, bool? ipv6) async =>
+            fakeServerSocket,
+      );
+      final int result = await portForwarder.forward(100);
+      expect(result, 200);
+
+      final FakeSocket fakeSocket1 = FakeSocket();
+      final FakeSocket fakeSocket2 = FakeSocket();
+      fakeServerSocket.controller.add(fakeSocket1);
+      fakeServerSocket.controller.add(fakeSocket2);
+
+      final Stream<DaemonMessage> broadcastOutput = serverDaemonConnection.incomingCommands.asBroadcastStream();
+
+      final DaemonMessage message1 = await broadcastOutput.first;
+
+      expect(message1.data['id'], isNotNull);
+      expect(message1.data['method'], 'proxy.connect');
+      expect(message1.data['params'], <String, Object?>{'port': 100});
+
+      const String id1 = 'random_id1';
+      serverDaemonConnection.sendResponse(message1.data['id']!, id1);
+
+      final DaemonMessage message2 = await broadcastOutput.first;
+
+      expect(message2.data['id'], isNotNull);
+      expect(message2.data['id'], isNot(message1.data['id']));
+      expect(message2.data['method'], 'proxy.connect');
+      expect(message2.data['params'], <String, Object?>{'port': 100});
+
+      const String id2 = 'random_id2';
+      serverDaemonConnection.sendResponse(message2.data['id']!, id2);
+
+      await pumpEventQueue();
+
+      // Closes the socket after port forwarder dispose.
+      expect(fakeSocket1.closeCalled, false);
+      expect(fakeSocket2.closeCalled, false);
+      await portForwarder.dispose();
+      expect(fakeSocket1.closeCalled, true);
+      expect(fakeSocket2.closeCalled, true);
+    });
   });
 
   final Map<String, Object> fakeDevice = <String, Object>{
@@ -316,6 +364,137 @@ void main() {
       expect(fakeFilter.devices![1].id, fakeDevice2['id']);
     });
   });
+
+  group('ProxiedDartDevelopmentService', () {
+    testWithoutContext('forwards start and shutdown to remote', () async {
+      final FakeProxiedPortForwarder portForwarder = FakeProxiedPortForwarder();
+      portForwarder.originalRemotePortReturnValue = 200;
+      portForwarder.forwardReturnValue = 400;
+      final ProxiedDartDevelopmentService dds = ProxiedDartDevelopmentService(
+        clientDaemonConnection,
+        'test_id',
+        logger: bufferLogger,
+        proxiedPortForwarder: portForwarder,
+      );
+
+      final Stream<DaemonMessage> broadcastOutput = serverDaemonConnection.incomingCommands.asBroadcastStream();
+
+      final Future<void> startFuture = dds.startDartDevelopmentService(
+        Uri.parse('http://127.0.0.1:100/fake'),
+        disableServiceAuthCodes: true,
+        hostPort: 150,
+        ipv6: false,
+        logger: bufferLogger,
+      );
+
+      final DaemonMessage startMessage = await broadcastOutput.first;
+      expect(startMessage.data['id'], isNotNull);
+      expect(startMessage.data['method'], 'device.startDartDevelopmentService');
+      expect(startMessage.data['params'], <String, Object?>{
+        'deviceId': 'test_id',
+        'vmServiceUri': 'http://127.0.0.1:200/fake',
+        'disableServiceAuthCodes': true,
+      });
+
+      serverDaemonConnection.sendResponse(startMessage.data['id']!, 'http://127.0.0.1:300/remote');
+
+      await startFuture;
+      expect(portForwarder.receivedLocalForwardedPort, 100);
+      expect(portForwarder.forwardedDevicePort, 300);
+      expect(portForwarder.forwardedHostPort, 150);
+      expect(portForwarder.forwardedIpv6, false);
+
+      expect(dds.uri, Uri.parse('http://127.0.0.1:400/remote'));
+
+      unawaited(dds.shutdown());
+
+      final DaemonMessage shutdownMessage = await broadcastOutput.first;
+      expect(shutdownMessage.data['id'], isNotNull);
+      expect(shutdownMessage.data['method'], 'device.shutdownDartDevelopmentService');
+    });
+
+    testWithoutContext('starts a local dds if the VM service port is not a forwarded port', () async {
+      final FakeProxiedPortForwarder portForwarder = FakeProxiedPortForwarder();
+      final FakeDartDevelopmentService localDds = FakeDartDevelopmentService();
+      localDds.uri = Uri.parse('http://127.0.0.1:450/local');
+      final ProxiedDartDevelopmentService dds = ProxiedDartDevelopmentService(
+        clientDaemonConnection,
+        'test_id',
+        logger: bufferLogger,
+        proxiedPortForwarder: portForwarder,
+        localDds: localDds,
+      );
+
+      expect(localDds.startCalled, false);
+      await dds.startDartDevelopmentService(
+        Uri.parse('http://127.0.0.1:100/fake'),
+        disableServiceAuthCodes: true,
+        hostPort: 150,
+        ipv6: false,
+        logger: bufferLogger,
+      );
+
+      expect(localDds.startCalled, true);
+      expect(portForwarder.receivedLocalForwardedPort, 100);
+      expect(portForwarder.forwardedDevicePort, null);
+
+      expect(dds.uri, Uri.parse('http://127.0.0.1:450/local'));
+
+      expect(localDds.shutdownCalled, false);
+      await dds.shutdown();
+      expect(localDds.shutdownCalled, true);
+
+      await serverDaemonConnection.dispose();
+      expect(await serverDaemonConnection.incomingCommands.isEmpty, true);
+    });
+
+    testWithoutContext('starts a local dds if the remote VM does not support starting DDS', () async {
+      final FakeProxiedPortForwarder portForwarder = FakeProxiedPortForwarder();
+      portForwarder.originalRemotePortReturnValue = 200;
+      final FakeDartDevelopmentService localDds = FakeDartDevelopmentService();
+      localDds.uri = Uri.parse('http://127.0.0.1:450/local');
+      final ProxiedDartDevelopmentService dds = ProxiedDartDevelopmentService(
+        clientDaemonConnection,
+        'test_id',
+        logger: bufferLogger,
+        proxiedPortForwarder: portForwarder,
+        localDds: localDds,
+      );
+
+      final Stream<DaemonMessage> broadcastOutput = serverDaemonConnection.incomingCommands.asBroadcastStream();
+
+      final Future<void> startFuture = dds.startDartDevelopmentService(
+        Uri.parse('http://127.0.0.1:100/fake'),
+        disableServiceAuthCodes: true,
+        hostPort: 150,
+        ipv6: false,
+        logger: bufferLogger,
+      );
+
+      expect(localDds.startCalled, false);
+      final DaemonMessage startMessage = await broadcastOutput.first;
+      expect(startMessage.data['id'], isNotNull);
+      expect(startMessage.data['method'], 'device.startDartDevelopmentService');
+      expect(startMessage.data['params'], <String, Object?>{
+        'deviceId': 'test_id',
+        'vmServiceUri': 'http://127.0.0.1:200/fake',
+        'disableServiceAuthCodes': true,
+      });
+
+      serverDaemonConnection.sendErrorResponse(startMessage.data['id']!, 'command not understood: device.startDartDevelopmentService', StackTrace.current);
+
+      await startFuture;
+      expect(localDds.startCalled, true);
+      expect(portForwarder.receivedLocalForwardedPort, 100);
+      expect(portForwarder.forwardedDevicePort, null);
+
+      expect(dds.uri, Uri.parse('http://127.0.0.1:450/local'));
+
+      expect(localDds.shutdownCalled, false);
+      await dds.shutdown();
+      expect(localDds.shutdownCalled, true);
+    });
+  });
 }
 
 class FakeDaemonStreams implements DaemonStreams {
@@ -392,6 +571,7 @@ class FakeSocket extends Fake implements Socket {
   @override
   Future<void> close() async {
     closeCalled = true;
+    doneCompleter.complete(true);
   }
 
   @override
@@ -441,3 +621,57 @@ class FakeDeviceDiscoveryFilter extends Fake implements DeviceDiscoveryFilter {
     return filteredDevices!;
   }
 }
+
+class FakeProxiedPortForwarder extends Fake implements ProxiedPortForwarder {
+  int? originalRemotePortReturnValue;
+  int? receivedLocalForwardedPort;
+
+  int? forwardReturnValue;
+  int? forwardedDevicePort;
+  int? forwardedHostPort;
+  bool? forwardedIpv6;
+
+  @override
+  int? originalRemotePort(int localForwardedPort) {
+    receivedLocalForwardedPort = localForwardedPort;
+    return originalRemotePortReturnValue;
+  }
+
+  @override
+  Future<int> forward(int devicePort, {int? hostPort, bool? ipv6}) async {
+    forwardedDevicePort = devicePort;
+    forwardedHostPort = hostPort;
+    forwardedIpv6 = ipv6;
+    return forwardReturnValue!;
+  }
+}
+
+class FakeDartDevelopmentService extends Fake implements DartDevelopmentService {
+  bool startCalled = false;
+  Uri? startUri;
+
+  bool shutdownCalled = false;
+
+  @override
+  Future<void> get done => _completer.future;
+  final Completer<void> _completer = Completer<void>();
+
+  @override
+  Uri? uri;
+
+  @override
+  Future<void> startDartDevelopmentService(
+    Uri vmServiceUri, {
+    required Logger logger,
+    int? hostPort,
+    bool? ipv6,
+    bool? disableServiceAuthCodes,
+    bool cacheStartupProfile = false,
+  }) async {
+    startCalled = true;
+    startUri = vmServiceUri;
+  }
+
+  @override
+  Future<void> shutdown() async => shutdownCalled = true;
+}