[ Service ] Add support for DDS to package:dart_runtime_service_vm

The DartRuntimeService based VM service implementation now has support
for launching DDS instances and responding to _yieldControlToDDS RPC
invocations from DDS instances.

package:vm_service test suite is ~97% passing with this change.

TEST=Local testing.
Change-Id: I2f2f1b0926845134578f08d073ed7606f1fc4173
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/490320
Reviewed-by: Jessy Yameogo <yjessy@google.com>
Reviewed-by: Nicholas Shahan <nshahan@google.com>

pkg/dart_runtime_service/lib/dart_runtime_service.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 export 'src/clients.dart';
+export 'src/dart_development_service_manager.dart';
 export 'src/dart_runtime_service.dart';
 export 'src/dart_runtime_service_backend.dart';
 export 'src/dart_runtime_service_options.dart';

pkg/dart_runtime_service/lib/src/clients.dart

@@ -25,6 +25,7 @@ base class Client {
     required UnmodifiableClientNamedLookup clients,
     required EventStreamMethods eventStreamMethods,
     required this.backend,
+    required this.artificial,
     String? name,
   }) {
     _name = name ?? defaultClientName;
@@ -43,6 +44,17 @@ base class Client {
   late final DartRuntimeServiceRpcs _internalRpcs;
   final DartRuntimeServiceBackend backend;
 
+  /// If `true`, this client was created via
+  /// [DartRuntimeService.addArtificialClient].
+  ///
+  /// [DartRuntimeServiceBackend]s sometimes need to be able to create clients
+  /// that aren't associated with an active connection to the service. For
+  /// example, the Dart VM provides native APIs to invoke service RPCs. This
+  /// can be implemented by manually creating a [StreamChannel] for native RPC
+  /// invocations to be added to, which is then used to create an artificial
+  /// client.
+  final bool artificial;
+
   /// The logger to be used when handling requests from this client.
   Logger get logger => Logger(toString());
 
@@ -177,39 +189,87 @@ base class Client {
   String toString() => 'Client ($name)';
 }
 
+/// An interface that allows for controlling whether or not new [Client]
+/// connections should be accepted or rejected.
+abstract interface class ClientConnectionController {
+  /// Accept connection requests from new [Client]s.
+  void acceptConnections();
+
+  /// Reject connection requests from new [Client]s, redirecting them to
+  /// connect to [redirectUri] instead.
+  void rejectConnections({required Uri redirectUri});
+}
+
 /// Used for keeping track and managing clients that are connected to a given
 /// service.
 ///
 /// Call [addClient] when a client connects to your service.
-base class ClientManager {
+base class ClientManager implements ClientConnectionController {
   ClientManager({required this.backend, required this.eventStreamMethods});
 
   static const _kServicePrologue = 's';
   final DartRuntimeServiceBackend backend;
   final EventStreamMethods eventStreamMethods;
 
+  final _logger = Logger('$ClientManager');
+
+  /// Returns `true` if new [Client] connections should be accepted.
+  ///
+  /// If `false`, [redirectUri] will be non-null and should be included in a
+  /// redirect response.
+  bool get acceptNewConnections => redirectUri == null;
+
+  /// The [Uri] pointing to the service that [Client]s should attempt to connect
+  /// to.
+  ///
+  /// Returns `null` if [acceptNewConnections] is `true`.
+  Uri? get redirectUri => _redirectUri;
+  Uri? _redirectUri;
+
   /// The set of [Client]s currently connected to the service.
   ///
   /// Each client is assigned a unique identifier, prefixed with
   /// [_kServicePrologue] (e.g., 's1'). This identifier is used when invoking
   /// service extensions registered by the client to indicate which client
   /// is responsible for handling the service extension invocation.
-  final clients = ClientNamedLookup(prefix: _kServicePrologue);
+  UnmodifiableClientNamedLookup get clients =>
+      UnmodifiableClientNamedLookup(_clients);
+  final _clients = ClientNamedLookup(prefix: _kServicePrologue);
+
+  @override
+  void acceptConnections() {
+    _redirectUri = null;
+    _logger.info('Accepting new connections.');
+  }
+
+  @override
+  void rejectConnections({required Uri redirectUri}) {
+    _redirectUri = redirectUri;
+    _logger.info(
+      'No longer accepting new connections. Redirecting connections to '
+      '$redirectUri.',
+    );
+  }
 
   /// Creates a [Client] from [connection] and adds it to the list of connected
   /// clients.
   ///
   /// This should be called when a client connects to the service.
   @mustCallSuper
-  Client addClient({required StreamChannel<String> connection, String? name}) {
+  Client addClient({
+    required StreamChannel<String> connection,
+    String? name,
+    bool artificial = false,
+  }) {
     final client = Client(
       connection: connection,
-      clients: UnmodifiableClientNamedLookup(clients),
+      clients: clients,
       eventStreamMethods: eventStreamMethods,
       backend: backend,
       name: name,
+      artificial: artificial,
     );
-    final namespace = clients.add(client);
+    final namespace = _clients.add(client);
     client.initialize(namespace: namespace).then((_) {
       // Remove the client from the clients list when it disconnects.
       removeClient(client);
@@ -224,8 +284,8 @@ base class ClientManager {
   @mustCallSuper
   @visibleForOverriding
   void removeClient(Client client) {
-    if (clients.contains(client)) {
-      clients.remove(client);
+    if (_clients.contains(client)) {
+      _clients.remove(client);
     }
   }
 
@@ -235,7 +295,7 @@ base class ClientManager {
     // Close all incoming websocket connections.
     final futures = <Future<void>>[];
     // Copy `clients` to guard against modification while iterating.
-    for (final client in clients.toList()) {
+    for (final client in _clients.toList()) {
       futures.add(
         Future.sync(() => removeClient(client)).whenComplete(client.close),
       );

pkg/dart_runtime_service/lib/src/dart_development_service_manager.dart

@@ -0,0 +1,201 @@
+// Copyright (c) 2026, the Dart project authors.  Please see the AUTHORS file
+// for details. 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:collection';
+
+import 'package:dds/dds.dart';
+import 'package:dds/dds_launcher.dart';
+import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+import 'package:logging/logging.dart';
+import 'package:vm_service/vm_service.dart';
+
+import '../dart_runtime_service.dart';
+
+/// Manages the lifecycle of the [DartDevelopmentService] (DDS).
+///
+/// Services can use this class to either launch their own DDS instance using
+/// [start], or wait for an external DDS connection (e.g., from Flutter Tools).
+final class DartDevelopmentServiceManager {
+  DartDevelopmentServiceManager({
+    required this.frontend,
+    required this.launchOnStart,
+    required this.host,
+    required this.port,
+  });
+
+  final DartRuntimeService frontend;
+
+  /// `true` if a DDS instance should be started immediately after the service
+  /// is initialized.
+  final bool launchOnStart;
+
+  /// The host DDS should attempt to bind to.
+  final String host;
+
+  /// The port DDS should attempt to bind to.
+  final int port;
+
+  static const _kUri = 'uri';
+
+  /// The HTTP URI pointing to a Dart Development Service (DDS) instance.
+  ///
+  /// If DDS is not running, [uri] returns null.
+  Uri? get uri => _launcher?.uri;
+
+  final _logger = Logger('$DartDevelopmentServiceManager');
+  DartDevelopmentServiceLauncher? _launcher;
+
+  /// The set of RPCs that must be registered for DDS to function.
+  late final rpcs = UnmodifiableListView<ServiceRpcHandler>([
+    ('_yieldControlToDDS', _yieldControlToDDS),
+  ]);
+
+  Future<Uri> get ddsConnected => _yieldCompleter.future;
+  var _yieldCompleter = Completer<Uri>();
+
+  /// Launches a Dart Development Service (DDS) instance that will attempt to
+  /// connect to the VM service at [vmServiceUri].
+  Future<void> start({required Uri vmServiceUri}) async {
+    assert(launchOnStart);
+    final ddsBindUri = Uri(scheme: 'http', host: host, port: port);
+    try {
+      _logger.info('Launching DDS at $ddsBindUri...');
+      _launcher = await DartDevelopmentServiceLauncher.start(
+        remoteVmServiceUri: vmServiceUri,
+        enableAuthCodes: !frontend.config.disableAuthCodes,
+        enableServicePortFallback: frontend.config.enableServicePortFallback,
+        serveDevTools: frontend.config.serveDevTools,
+        serviceUri: ddsBindUri,
+      );
+      unawaited(_launcher!.done.then((_) => _cleanup()));
+      _logger.info('DDS is served at $uri');
+    } on ExistingDartDevelopmentServiceException catch (e) {
+      _logger.warning('A DDS instance already exists at ${e.ddsUri}.');
+    } on DartDevelopmentServiceException catch (e) {
+      switch (e.errorCode) {
+        case DartDevelopmentServiceException.connectionError:
+          _logger.warning('Failed to connect to the VM service: ${e.message}.');
+        case DartDevelopmentServiceException.failedToStartError:
+          _logger.warning('Failed to start DDS: ${e.message}');
+      }
+    }
+  }
+
+  /// Shuts down the Dart Development Service (DDS) instance, if it exists.
+  Future<void> shutdown() async {
+    if (_launcher == null) {
+      return;
+    }
+    _logger.info('Shutting down DDS...');
+    await _launcher?.shutdown();
+    _cleanup();
+    _logger.info('DDS shutdown.');
+  }
+
+  void _cleanup() {
+    _launcher = null;
+    _yieldCompleter = Completer<Uri>();
+  }
+
+  /// Invoked by DDS when it connects to the service to ensure that it's the
+  /// only direct client of the service.
+  ///
+  /// DDS must be the only client as it takes over some of the responsibilities
+  /// of the VM service, such as client-registered service extension routing,
+  /// stream management, etc. In order for DDS to make some assumptions about
+  /// the state of the service, all other clients must connect to the service
+  /// through DDS.
+  ///
+  /// When invoked, new client connections to the service are disabled, with
+  /// redirect responses pointing to the DDS instance sent when connections are
+  /// attempted. An event is sent on the `Service` stream to each non-DDS
+  /// client explaining why they're about to be disconnected before the service
+  /// closes the client's connection.
+  ///
+  /// If the DDS client disconnects, the service will once again allow for
+  /// direct connections.
+  Future<RpcResponse> _yieldControlToDDS(
+    json_rpc.Parameters params,
+    Client client,
+  ) async {
+    var uri = _launcher?.uri;
+    if (uri != null) {
+      RpcException.featureDisabled.throwException(
+        data: {
+          'ddsUri': uri,
+          'details': 'A DDS instance is already connected at $uri.',
+        },
+      );
+    }
+    uri = Uri.tryParse(params[_kUri].asString);
+    if (uri == null) {
+      RpcException.invalidParams.throwExceptionWithDetails(
+        details: "'$_kUri' is not a valid URI.",
+      );
+    }
+    _logger.info(
+      'Rejecting future connections and disconnecting non-DDS clients.',
+    );
+    frontend.clientConnectionController.rejectConnections(redirectUri: uri);
+
+    // Register a callback to cleanup state if DDS disconnects.
+    unawaited(
+      client.done.then((_) async {
+        await _yieldCompleter.future;
+        _cleanup();
+        frontend.clientConnectionController.acceptConnections();
+        _logger.info('DDS disconnected. Accepting future connections.');
+      }),
+    );
+
+    client.setName('DDS');
+
+    // Notify clients why they're being disconnected from the VM service.
+    _DartDevelopmentServiceConnectedEvent(
+      uri: uri,
+    ).send(eventStreamMethods: frontend.eventStreams, excludedClient: client);
+
+    await Future.wait([
+      for (final client in frontend.clients.toList().where(
+        (e) => e != client && !e.artificial,
+      ))
+        client.close(),
+    ]);
+
+    _logger.info('Non-DDS clients disconnected.');
+    _yieldCompleter.complete(uri);
+    return Success().toJson();
+  }
+}
+
+/// An event notifying [Client]s that DDS has connected and their connection to
+/// the service is about to be closed.
+final class _DartDevelopmentServiceConnectedEvent extends StreamEvent {
+  _DartDevelopmentServiceConnectedEvent({required this.uri})
+    : super(
+        streamId: EventStreams.kService,
+        kind: 'DartDevelopmentServiceConnected',
+      );
+
+  final Uri uri;
+
+  static const _kMessage = 'message';
+
+  @override
+  Map<String, Object?> toJson() => {
+    StreamEvent.kStreamId: streamId,
+    StreamEvent.kEvent: {
+      ...Event(
+        kind: kind,
+        timestamp: DateTime.now().millisecondsSinceEpoch,
+      ).toJson(),
+      _kMessage:
+          'A Dart Developer Service instance has connected and this direct '
+          'connection to the VM service will now be closed. Please reconnect '
+          'to the Dart Development Service at $uri.',
+      DartDevelopmentServiceManager._kUri: uri.toString(),
+    },
+  };
+}

pkg/dart_runtime_service/lib/src/dart_runtime_service.dart

@@ -103,14 +103,20 @@ class DartRuntimeService {
 
   final _logger = Logger('$DartRuntimeService');
 
+  /// Exposes methods for controlling acceptance of new [Client] connections.
+  ClientConnectionController get clientConnectionController => clientManager;
+
   @visibleForTesting
   late final ClientManager clientManager = ClientManager(
     backend: backend,
     eventStreamMethods: eventStreamManager,
   );
 
-  UnmodifiableClientNamedLookup get clients =>
-      UnmodifiableClientNamedLookup(clientManager.clients);
+  /// The set of currently connected [Client]s.
+  UnmodifiableClientNamedLookup get clients => clientManager.clients;
+
+  /// Exposes methods for interacting with event streams.
+  EventStreamMethods get eventStreams => eventStreamManager;
 
   @visibleForTesting
   late final eventStreamManager = EventStreamManager(
@@ -181,7 +187,11 @@ class DartRuntimeService {
     required StreamChannel<String> connection,
     required String name,
   }) {
-    return clientManager.addClient(connection: connection, name: name);
+    return clientManager.addClient(
+      connection: connection,
+      name: name,
+      artificial: true,
+    );
   }
 
   Future<void> _startServer() async {

pkg/dart_runtime_service/lib/src/dart_runtime_service_options.dart

@@ -13,6 +13,8 @@ class DartRuntimeServiceOptions {
     this.disableOriginCheck = false,
     this.sseHandlerPath,
     this.autoStart = true,
+    this.serveDevTools = false,
+    this.enableServicePortFallback = false,
   });
 
   /// If true, enables log output for the service.
@@ -43,6 +45,13 @@ class DartRuntimeServiceOptions {
   /// If true, the HTTP server will be started on initialization.
   final bool autoStart;
 
+  /// If true, Dart DevTools should be made available via the HTTP server.
+  final bool serveDevTools;
+
+  /// If true, the service should attempt to bind to a different port if [port]
+  /// is unavailable.
+  final bool enableServicePortFallback;
+
   DartRuntimeServiceOptions copyWith({
     bool? enableLogging,
     int? port,
@@ -50,6 +59,8 @@ class DartRuntimeServiceOptions {
     bool? disableOriginCheck,
     String? sseHandlerPath,
     bool? autoStart,
+    bool? serveDevTools,
+    bool? enableServicePortFallback,
   }) {
     return DartRuntimeServiceOptions(
       enableLogging: enableLogging ?? this.enableLogging,
@@ -58,6 +69,9 @@ class DartRuntimeServiceOptions {
       disableOriginCheck: disableOriginCheck ?? this.disableOriginCheck,
       sseHandlerPath: sseHandlerPath ?? this.sseHandlerPath,
       autoStart: autoStart ?? this.autoStart,
+      serveDevTools: serveDevTools ?? this.serveDevTools,
+      enableServicePortFallback:
+          enableServicePortFallback ?? this.enableServicePortFallback,
     );
   }
 }

pkg/dart_runtime_service/lib/src/dart_runtime_service_rpcs.dart

@@ -20,6 +20,8 @@ import 'utils.dart';
 typedef RpcHandlerWithNoParameters = FutureOr<RpcResponse> Function();
 typedef RpcHandlerWithParameters =
     FutureOr<RpcResponse> Function(json_rpc.Parameters);
+typedef RpcHandlerWithParametersAndClient =
+    FutureOr<RpcResponse> Function(json_rpc.Parameters, Client);
 
 typedef ServiceRpcHandler = (String, Function?);
 
@@ -70,19 +72,17 @@ final class DartRuntimeServiceRpcs {
     _backendFallbacks.addAll(backend.fallbacks);
   }
 
-  void addBackendFallbacks({
-    required List<RpcHandlerWithParameters> fallbacks,
-  }) => _backendFallbacks.addAll(fallbacks);
-
   /// Registers the set of platform-agnostic and backend RPCs for use by
   /// [client].
   void registerRpcsWithPeer(json_rpc.Peer clientPeer) {
     for (final (method, callback) in [..._commonRpcs, ..._backendRpcs]) {
       if (callback == null) continue;
       if (callback is! RpcHandlerWithNoParameters &&
-          callback is! RpcHandlerWithParameters) {
+          callback is! RpcHandlerWithParameters &&
+          callback is! RpcHandlerWithParametersAndClient) {
         throw StateError("Callback for '$method' is not valid. ($callback).");
       }
+      client.logger.info('Registering $method');
       clientPeer.registerMethod(method, (json_rpc.Parameters parameters) async {
         try {
           late RpcResponse response;
@@ -94,6 +94,10 @@ final class DartRuntimeServiceRpcs {
             client.logger.info('Invoked $method (${parameters.value})');
             response = await callback(parameters);
             client.logger.info('Response: $response');
+          } else if (callback is RpcHandlerWithParametersAndClient) {
+            client.logger.info('Invoked $method (${parameters.value})');
+            response = await callback(parameters, client);
+            client.logger.info('Response: $response');
           }
           return response;
         } catch (e, st) {

pkg/dart_runtime_service/lib/src/event_streams.dart

@@ -26,7 +26,7 @@ abstract base class StreamEvent {
   final int timestamp = DateTime.now().millisecondsSinceEpoch;
 
   void send({
-    required EventStreamManager eventStreamMethods,
+    required EventStreamMethods eventStreamMethods,
     Client? excludedClient,
   }) {
     eventStreamMethods.streamNotify(
@@ -161,17 +161,26 @@ class EventStreamManager implements EventStreamMethods {
     final streamLogger = Logger('${_logger.name} ($streamId)');
     if (streamListeners.containsKey(streamId)) {
       final listeners = streamListeners[streamId]!;
-      String eventString;
-      if (data is Uint8List) {
-        eventString = '<binary data>';
-      } else if (data is StreamEvent) {
-        eventString = data.toJson().toString();
-      } else {
-        eventString = '<unknown>';
+      // Don't log event string for streams known to send large amounts of
+      // data.
+      if (!const {
+        EventStreams.kStdout,
+        EventStreams.kStderr,
+        EventStreams.kHeapSnapshot,
+        EventStreams.kLogging,
+      }.contains(streamId)) {
+        String eventString;
+        if (data is Uint8List) {
+          eventString = '<binary data>';
+        } else if (data is StreamEvent) {
+          eventString = data.toJson().toString();
+        } else {
+          eventString = '<unknown>';
+        }
+        streamLogger.info(
+          'Sending event to ${listeners.length} clients: $eventString.',
+        );
       }
-      streamLogger.info(
-        'Sending event to ${listeners.length} clients: $eventString',
-      );
 
       for (final listener in listeners) {
         if (listener == excludedClient) {

pkg/dart_runtime_service/lib/src/handlers.dart

@@ -18,6 +18,10 @@ import 'clients.dart';
 import 'dart_runtime_service.dart';
 import 'dart_runtime_service_backend.dart';
 
+/// Return from a handler to indicate that the request can't be handled by the
+/// current handler.
+Response notHandledByHandler() => Response.notFound('');
+
 /// Creates [Middleware] responsible for logging the result of HTTP requests.
 ///
 /// Note: this only outputs logs when the response is sent. Connections that
@@ -161,14 +165,34 @@ Handler httpRequestHandler({required DartRuntimeService frontend}) =>
     };
 
 /// Creates a [Handler] for incoming web socket connections.
-Handler webSocketClientHandler({required ClientManager clientManager}) =>
-    webSocketHandler((WebSocketChannel ws, _) {
-      // Note: the WebSocketChannel type below is needed for compatibility with
-      // package:shelf_web_socket v2.
-      final logger = Logger('WebSocketHandler');
-      logger.info('New web socket connection. Creating $Client.');
-      clientManager.addClient(connection: ws.cast<String>());
-    });
+Handler webSocketClientHandler({required ClientManager clientManager}) {
+  final logger = Logger('WebSocketHandler');
+  // Note: the WebSocketChannel type below is needed for compatibility with
+  // package:shelf_web_socket v2.
+  final handler = webSocketHandler((WebSocketChannel ws, _) {
+    logger.info('New web socket connection. Creating $Client.');
+    clientManager.addClient(connection: ws.cast<String>());
+  });
+
+  return (request) {
+    if (!request.isWebSocketUpgradeRequest) {
+      return notHandledByHandler();
+    }
+    if (!clientManager.acceptNewConnections) {
+      logger.info(
+        'New connections not accepted. Rejecting web socket connection.',
+      );
+      final redirectUri = clientManager.redirectUri;
+      if (redirectUri != null) {
+        return Response.seeOther(clientManager.redirectUri.toString());
+      }
+      return Response.forbidden(
+        'New connections not accepted. Rejecting web socket connection.',
+      );
+    }
+    return handler(request);
+  };
+}
 
 /// Creates a [Handler] for incoming SSE connections.
 Handler sseClientHandler({
@@ -176,7 +200,6 @@ Handler sseClientHandler({
   required String sseHandlerPath,
   required String? authCode,
 }) {
-  final logger = Logger('SSEClientHandler');
   // Give connections time to reestablish before considering them closed.
   // Required to reestablish connections killed by UberProxy.
   const sseKeepAlive = Duration(seconds: 30);
@@ -186,10 +209,35 @@ Handler sseClientHandler({
     keepAlive: sseKeepAlive,
   );
 
-  handler.connections.rest.listen((sseConnection) {
-    logger.info('New SSE connection. Creating $Client.');
-    clientManager.addClient(connection: sseConnection);
-  });
+  final logger = Logger('SSEClientHandler');
 
-  return handler.handler;
+  return (request) {
+    if (!clientManager.acceptNewConnections && request.isSSEConnectionRequest) {
+      logger.info('New connections not accepted. Rejecting SSE connection.');
+      final redirectUri = clientManager.redirectUri;
+      if (redirectUri != null) {
+        return Response.seeOther(clientManager.redirectUri.toString());
+      }
+      return Response.forbidden(
+        'New connections not accepted. Rejecting SSE connection.',
+      );
+    }
+
+    handler.connections.rest.listen((sseConnection) {
+      logger.info('New SSE connection. Creating $Client.');
+      clientManager.addClient(connection: sseConnection);
+    });
+
+    return handler.handler(request);
+  };
+}
+
+/// Adds checks for specific headers to determine if a [Request] is attempting
+/// to establish a web socket or SSE connection.
+extension on Request {
+  bool get isWebSocketUpgradeRequest =>
+      headers.containsKey('Sec-WebSocket-Key');
+
+  bool get isSSEConnectionRequest =>
+      headers['accept'] == 'text/event-stream' && method == 'GET';
 }

pkg/dart_runtime_service/pubspec.yaml

@@ -9,6 +9,7 @@ resolution: workspace
 
 # Use 'any' constraints here; we get our versions from the DEPS file.
 dependencies:
+  dds: any
   dds_service_extensions: any
   json_rpc_2: any
   logging: any

pkg/dart_runtime_service_vm/bin/vm_service_entrypoint.dart

@@ -18,12 +18,10 @@ const entrypoint = pragma(
 
 // The TCP IP that DDS listens on.
 @entrypoint
-// ignore: unused_element
 String _ddsIP = '';
 
 // The TCP port that DDS listens on.
 @entrypoint
-// ignore: unused_element
 int _ddsPort = 0;
 
 // The TCP port that the HTTP server listens on.
@@ -45,7 +43,6 @@ bool _authCodesDisabled = false;
 
 // Should the HTTP server run in devmode?
 @entrypoint
-// ignore: unused_element
 bool _originCheckDisabled = false;
 
 // Location of file to output VM service connection info.
@@ -82,15 +79,12 @@ void _registerIsolate(int portId, SendPort sendPort, String name) =>
 StreamSubscription<ProcessSignal>? _signalSubscription;
 
 @entrypoint
-// ignore: unused_element
 bool _serveDevtools = true;
 
 @entrypoint
-// ignore: unused_element
 bool _enableServicePortFallback = false;
 
 @entrypoint
-// ignore: unused_element
 bool _waitForDdsToAdvertiseService = false;
 
 @entrypoint
@@ -124,11 +118,19 @@ Future<void> main([List<String> args = const []]) async {
       disableAuthCodes: _authCodesDisabled,
       disableOriginCheck: _originCheckDisabled,
       autoStart: _autoStart,
+      serveDevTools: _serveDevtools,
+      enableServicePortFallback: _enableServicePortFallback,
     ),
     backendBuilder: (frontend) => DartRuntimeServiceVMBackend(
       frontend: frontend,
       signalWatch: _signalWatch!,
       runningIsolatesStream: _isolateRegistrationStreamController.stream,
+      ddsManager: DartDevelopmentServiceManager(
+        frontend: frontend,
+        launchOnStart: _waitForDdsToAdvertiseService,
+        host: _ddsIP,
+        port: _ddsPort,
+      ),
     ),
   );
 }

pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart

@@ -31,6 +31,7 @@ class DartRuntimeServiceVMBackend
     required super.frontend,
     required this.signalWatch,
     required Stream<VmRunningIsolate> runningIsolatesStream,
+    required this._ddsManager,
   }) : isolateManager = VmIsolateManager(
          runningIsolatesStream: runningIsolatesStream,
        );
@@ -79,9 +80,16 @@ class DartRuntimeServiceVMBackend
 
   final _vmServiceRpcs = DartRuntimeServiceVmRpcs();
 
+  /// Adds support for launching and accepting connections from the
+  /// Dart Development Service.
+  final DartDevelopmentServiceManager _ddsManager;
+
   @override
-  UnmodifiableListView<ServiceRpcHandler> get rpcs =>
-      UnmodifiableListView([..._vmServiceRpcs.rpcs, ..._devFs.rpcs]);
+  UnmodifiableListView<ServiceRpcHandler> get rpcs => UnmodifiableListView([
+    ..._vmServiceRpcs.rpcs,
+    ..._devFs.rpcs,
+    ..._ddsManager.rpcs,
+  ]);
 
   @override
   UnmodifiableListView<RpcHandlerWithParameters>
@@ -118,6 +126,7 @@ class DartRuntimeServiceVMBackend
   @override
   Future<void> shutdown() async {
     _logger.info('Shutting down...');
+    await _ddsManager.shutdown();
     await Future.wait([
       _sigquitSubscription?.cancel() ?? Future<void>.value(),
       _nativeRpcClientStreamChannelController.local.sink.close(),
@@ -145,7 +154,10 @@ class DartRuntimeServiceVMBackend
     required Uri httpUri,
     required Uri wsUri,
   }) async {
-    // TODO(bkonyi): handle DDS connection case.
+    if (_ddsManager.launchOnStart) {
+      await _ddsManager.start(vmServiceUri: httpUri);
+      httpUri = await _ddsManager.ddsConnected;
+    }
     stdout.writeln('The Dart VM service is listening on $httpUri/');
     _nativeBindings.onServerAddressChange(httpUri.toString());
   }
@@ -173,7 +185,12 @@ class DartRuntimeServiceVMBackend
   /// Sends service requests to the Dart VM runtime for processing.
   Future<RpcResponse> sendToRuntime(json_rpc.Parameters request) async {
     final method = request.method;
-    final params = request.asMap.cast<String, Object?>();
+    // It's possible that a client will omit the parameters map for RPCs with
+    // no parameters. Don't try and cast the request unless the value is
+    // actually a map, otherwise assume there's no arugments.
+    final params = request.value is Map
+        ? request.asMap.cast<String, Object?>()
+        : const <String, Object?>{};
     if (params case {'isolateId': final String _}) {
       _logger.info(
         'Sending request to isolate. Method: $method Params: $params',
@@ -194,7 +211,7 @@ class DartRuntimeServiceVMBackend
   ///     Service.toggleWebServer())
   ///   - Isolate startup and shutdown notifications
   void _vmMessageHandler(List<Object?> message) {
-    _logger.info('VM message: $message');
+    _logger.fine('VM message: $message');
     switch (message) {
       case [final String streamId, final String eventJsonString]:
         // This is an event.

pkg/dds/test/dds_disconnects_existing_clients_test.dart

@@ -30,6 +30,7 @@ final tests = <VMTest>[
       expect(event.json!['uri'], dds.uri.toString());
       serviceMessageCompleter.complete();
     });
+    await service.streamListen(EventStreams.kService);
 
     // Start DDS, which should result in the original VM service client being
     // disconnected from the VM service.

runtime/bin/main_impl.cc

@@ -1436,14 +1436,15 @@ void main(int argc, char** argv) {
     }
   }
 
-  // Terminate process exit-code handler.
-  Process::TerminateExitCodeHandler();
-
   error = Dart_Cleanup();
   if (error != nullptr) {
     Syslog::PrintErr("VM cleanup failed: %s\n", error);
     free(error);
   }
+
+  // Terminate process exit-code handler.
+  Process::TerminateExitCodeHandler();
+
   const intptr_t global_exit_code = Process::GlobalExitCode();
   dart::embedder::Cleanup();
 

runtime/bin/vmservice_impl.cc

@@ -182,12 +182,6 @@ bool VmService::Setup(const char* server_ip,
     // port when the HTTP server is started.
     server_port = 0;
   }
-#if defined(EXPERIMENTAL_VM_SERVICE)
-  if (enable_experimental_vm_service) {
-    // TODO(bkonyi): remove once DDS support is added.
-    wait_for_dds_to_advertise_service = false;
-  }
-#endif
   if (wait_for_dds_to_advertise_service) {
     result = DartUtils::SetStringField(library, "_ddsIP", server_ip);
     SHUTDOWN_ON_ERROR(result);