diff --git a/pkg/dart_runtime_service/lib/dart_runtime_service.dart b/pkg/dart_runtime_service/lib/dart_runtime_service.dart index 9cd7815a2ae..359de5e0587 100644 --- a/pkg/dart_runtime_service/lib/dart_runtime_service.dart +++ b/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'; diff --git a/pkg/dart_runtime_service/lib/src/clients.dart b/pkg/dart_runtime_service/lib/src/clients.dart index fc1768adf5f..a055a880446 100644 --- a/pkg/dart_runtime_service/lib/src/clients.dart +++ b/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 connection, String? name}) { + Client addClient({ + required StreamChannel 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 = >[]; // 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), ); diff --git a/pkg/dart_runtime_service/lib/src/dart_development_service_manager.dart b/pkg/dart_runtime_service/lib/src/dart_development_service_manager.dart new file mode 100644 index 00000000000..6b5ba8ab7df --- /dev/null +++ b/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([ + ('_yieldControlToDDS', _yieldControlToDDS), + ]); + + Future get ddsConnected => _yieldCompleter.future; + var _yieldCompleter = Completer(); + + /// Launches a Dart Development Service (DDS) instance that will attempt to + /// connect to the VM service at [vmServiceUri]. + Future 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 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(); + } + + /// 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 _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 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(), + }, + }; +} diff --git a/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart b/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart index cdc4abfdfac..1ba64ddac83 100644 --- a/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart +++ b/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 connection, required String name, }) { - return clientManager.addClient(connection: connection, name: name); + return clientManager.addClient( + connection: connection, + name: name, + artificial: true, + ); } Future _startServer() async { diff --git a/pkg/dart_runtime_service/lib/src/dart_runtime_service_options.dart b/pkg/dart_runtime_service/lib/src/dart_runtime_service_options.dart index a90e78ff329..b6ea36cd46d 100644 --- a/pkg/dart_runtime_service/lib/src/dart_runtime_service_options.dart +++ b/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, ); } } diff --git a/pkg/dart_runtime_service/lib/src/dart_runtime_service_rpcs.dart b/pkg/dart_runtime_service/lib/src/dart_runtime_service_rpcs.dart index c357a60cdfb..97ff5b19b39 100644 --- a/pkg/dart_runtime_service/lib/src/dart_runtime_service_rpcs.dart +++ b/pkg/dart_runtime_service/lib/src/dart_runtime_service_rpcs.dart @@ -20,6 +20,8 @@ import 'utils.dart'; typedef RpcHandlerWithNoParameters = FutureOr Function(); typedef RpcHandlerWithParameters = FutureOr Function(json_rpc.Parameters); +typedef RpcHandlerWithParametersAndClient = + FutureOr Function(json_rpc.Parameters, Client); typedef ServiceRpcHandler = (String, Function?); @@ -70,19 +72,17 @@ final class DartRuntimeServiceRpcs { _backendFallbacks.addAll(backend.fallbacks); } - void addBackendFallbacks({ - required List 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) { diff --git a/pkg/dart_runtime_service/lib/src/event_streams.dart b/pkg/dart_runtime_service/lib/src/event_streams.dart index 266e1b9edd8..e15bfe609df 100644 --- a/pkg/dart_runtime_service/lib/src/event_streams.dart +++ b/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 = ''; - } else if (data is StreamEvent) { - eventString = data.toJson().toString(); - } else { - eventString = ''; + // 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 = ''; + } else if (data is StreamEvent) { + eventString = data.toJson().toString(); + } else { + eventString = ''; + } + 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) { diff --git a/pkg/dart_runtime_service/lib/src/handlers.dart b/pkg/dart_runtime_service/lib/src/handlers.dart index 84541cea7d5..eb0b32f00bd 100644 --- a/pkg/dart_runtime_service/lib/src/handlers.dart +++ b/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()); - }); +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()); + }); + + 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'; } diff --git a/pkg/dart_runtime_service/pubspec.yaml b/pkg/dart_runtime_service/pubspec.yaml index 4488548cf79..6db15a86ced 100644 --- a/pkg/dart_runtime_service/pubspec.yaml +++ b/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 diff --git a/pkg/dart_runtime_service_vm/bin/vm_service_entrypoint.dart b/pkg/dart_runtime_service_vm/bin/vm_service_entrypoint.dart index e6a3af2724d..2410356c4c4 100644 --- a/pkg/dart_runtime_service_vm/bin/vm_service_entrypoint.dart +++ b/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? _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 main([List 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, + ), ), ); } diff --git a/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart b/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart index bbf1e46b08a..f066bc27a62 100644 --- a/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart +++ b/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 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 get rpcs => - UnmodifiableListView([..._vmServiceRpcs.rpcs, ..._devFs.rpcs]); + UnmodifiableListView get rpcs => UnmodifiableListView([ + ..._vmServiceRpcs.rpcs, + ..._devFs.rpcs, + ..._ddsManager.rpcs, + ]); @override UnmodifiableListView @@ -118,6 +126,7 @@ class DartRuntimeServiceVMBackend @override Future shutdown() async { _logger.info('Shutting down...'); + await _ddsManager.shutdown(); await Future.wait([ _sigquitSubscription?.cancel() ?? Future.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 sendToRuntime(json_rpc.Parameters request) async { final method = request.method; - final params = request.asMap.cast(); + // 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() + : const {}; 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 message) { - _logger.info('VM message: $message'); + _logger.fine('VM message: $message'); switch (message) { case [final String streamId, final String eventJsonString]: // This is an event. diff --git a/pkg/dds/test/dds_disconnects_existing_clients_test.dart b/pkg/dds/test/dds_disconnects_existing_clients_test.dart index 5204b6ca09f..852088f77b1 100644 --- a/pkg/dds/test/dds_disconnects_existing_clients_test.dart +++ b/pkg/dds/test/dds_disconnects_existing_clients_test.dart @@ -30,6 +30,7 @@ final tests = [ 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. diff --git a/runtime/bin/main_impl.cc b/runtime/bin/main_impl.cc index e9a8add77dd..7d2a3a61558 100644 --- a/runtime/bin/main_impl.cc +++ b/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(); diff --git a/runtime/bin/vmservice_impl.cc b/runtime/bin/vmservice_impl.cc index a99635ae32a..048097db168 100644 --- a/runtime/bin/vmservice_impl.cc +++ b/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);