DTDManager handles the subscription to the service stream

Author: elliette

packages/devtools_app/lib/src/shared/editor/editor_client.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:devtools_app_shared/service.dart';
 import 'package:devtools_app_shared/utils.dart';
 import 'package:devtools_shared/devtools_shared.dart';
 import 'package:dtd/dtd.dart';
@@ -20,32 +21,28 @@ import 'api_classes.dart';
 /// ensure they are not breaking changes to already-shipped editors.
 class EditorClient extends DisposableController
     with AutoDisposeControllerMixin {
-  EditorClient(this._dtd) {
+  EditorClient(this._dtdManager) {
     unawaited(initialized); // Trigger async initialization.
   }
 
-  final DartToolingDaemon _dtd;
+  final DTDManager _dtdManager;
   late final initialized = _initialize();
 
+  DartToolingDaemon get _dtd => _dtdManager.connection.value!;
+
   String get gaId => EditorSidebar.id;
 
   Future<void> _initialize() async {
     autoDisposeStreamSubscription(
-      _dtd.onEvent(CoreDtdServiceConstants.servicesStreamId).listen((data) {
-        final kind = data.kind;
-        if (kind != CoreDtdServiceConstants.serviceRegisteredKind &&
-            kind != CoreDtdServiceConstants.serviceUnregisteredKind) {
-          return;
-        }
-
+      _dtdManager.serviceRegistrationBroadcastStream.listen((data) {
         final service = data.data[DtdParameters.service] as String?;
         if (service == null ||
             (service != editorServiceName && service != lspServiceName)) {
           return;
         }
 
         final isRegistered =
-            kind == CoreDtdServiceConstants.serviceRegisteredKind;
+            data.kind == CoreDtdServiceConstants.serviceRegisteredKind;
         final method = data.data[DtdParameters.method] as String;
         final capabilities =
             data.data[DtdParameters.capabilities] as Map<String, Object?>?;
@@ -127,7 +124,6 @@ class EditorClient extends DisposableController
       }),
     );
     await [
-      _dtd.streamListen(CoreDtdServiceConstants.servicesStreamId),
       _dtd.streamListen(editorStreamName).catchError((_) {
         // Because we currently call streamListen in two places (here and
         // ThemeManager) this can fail. It doesn't matter if this happens,

packages/devtools_app/lib/src/standalone_ui/ide_shared/property_editor/property_editor_panel.dart

@@ -4,9 +4,9 @@
 
 import 'dart:async';
 
+import 'package:devtools_app_shared/service.dart';
 import 'package:devtools_app_shared/ui.dart';
 import 'package:devtools_app_shared/utils.dart';
-import 'package:dtd/dtd.dart';
 import 'package:flutter/material.dart';
 
 import '../../../framework/scaffold/report_feedback_button.dart';
@@ -20,9 +20,9 @@ import 'property_editor_view.dart';
 
 /// The side panel for the Property Editor.
 class PropertyEditorPanel extends StatefulWidget {
-  const PropertyEditorPanel(this.dtd, {super.key});
+  const PropertyEditorPanel(this.dtdManager, {super.key});
 
-  final DartToolingDaemon dtd;
+  final DTDManager dtdManager;
 
   @override
   State<PropertyEditorPanel> createState() => _PropertyEditorPanelState();
@@ -38,7 +38,7 @@ class _PropertyEditorPanelState extends State<PropertyEditorPanel> {
   void initState() {
     super.initState();
 
-    final editor = EditorClient(widget.dtd);
+    final editor = EditorClient(widget.dtdManager);
     ga.screen(gac.PropertyEditorSidebar.id);
     unawaited(
       _editor = editor.initialized.then((_) {

packages/devtools_app/lib/src/standalone_ui/standalone_screen.dart

@@ -37,11 +37,11 @@ enum StandaloneScreenType {
       ),
       StandaloneScreenType.editorSidebar => _DtdConnectedScreen(
         dtdManager: dtdManager,
-        builder: (dtd) => EditorSidebarPanel(dtd),
+        builder: (dtdManager) => EditorSidebarPanel(dtdManager),
       ),
       StandaloneScreenType.propertyEditor => _DtdConnectedScreen(
         dtdManager: dtdManager,
-        builder: (dtd) => PropertyEditorPanel(dtd),
+        builder: (dtdManager) => PropertyEditorPanel(dtdManager),
       ),
     };
   }
@@ -58,7 +58,7 @@ class _DtdConnectedScreen extends StatelessWidget {
   const _DtdConnectedScreen({required this.dtdManager, required this.builder});
 
   final DTDManager dtdManager;
-  final Widget Function(DartToolingDaemon) builder;
+  final Widget Function(DTDManager) builder;
 
   @override
   Widget build(BuildContext context) {
@@ -80,7 +80,7 @@ class _DtdConnectedScreen extends StatelessWidget {
                   // reconnect occurs.
                   KeyedSubtree(
                     key: ValueKey(connection),
-                    child: builder(connection),
+                    child: builder(dtdManager),
                   ),
                 if (connectionState is! ConnectedDTDState)
                   NotConnectedOverlay(connectionState),

packages/devtools_app/lib/src/standalone_ui/vs_code/flutter_panel.dart

@@ -4,9 +4,9 @@
 
 import 'dart:async';
 
+import 'package:devtools_app_shared/service.dart';
 import 'package:devtools_app_shared/ui.dart';
 import 'package:devtools_app_shared/utils.dart';
-import 'package:dtd/dtd.dart';
 import 'package:flutter/material.dart';
 
 import '../../shared/analytics/analytics.dart' as ga;
@@ -22,9 +22,9 @@ import 'devtools/devtools_view.dart';
 /// Provides some basic functionality to improve discoverability of features
 /// such as creation of new projects, device selection and DevTools features.
 class EditorSidebarPanel extends StatefulWidget {
-  const EditorSidebarPanel(this.dtd, {super.key});
+  const EditorSidebarPanel(this.dtdManager, {super.key});
 
-  final DartToolingDaemon dtd;
+  final DTDManager dtdManager;
 
   @override
   State<EditorSidebarPanel> createState() => _EditorSidebarPanelState();
@@ -39,7 +39,7 @@ class _EditorSidebarPanelState extends State<EditorSidebarPanel> {
   void initState() {
     super.initState();
 
-    final editor = EditorClient(widget.dtd);
+    final editor = EditorClient(widget.dtdManager);
     ga.screen(editor.gaId);
     unawaited(_editor = editor.initialized.then((_) => editor));
   }

packages/devtools_app/test/shared/editor/editor_client_test.dart

@@ -8,10 +8,12 @@ import 'package:devtools_app/src/shared/editor/api_classes.dart';
 import 'package:devtools_app/src/shared/editor/editor_client.dart';
 import 'package:devtools_test/devtools_test.dart';
 import 'package:dtd/dtd.dart';
+import 'package:flutter/foundation.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:mockito/mockito.dart';
 
 void main() {
+  late MockDTDManager mockDTDManager;
   late MockDartToolingDaemon mockDtd;
   late EditorClient editorClient;
 
@@ -23,7 +25,9 @@ void main() {
   };
 
   setUp(() {
+    mockDTDManager = MockDTDManager();
     mockDtd = MockDartToolingDaemon();
+    when(mockDTDManager.connection).thenReturn(ValueNotifier(mockDtd));
 
     for (final MapEntry(key: method, value: responseJson)
         in methodToResponseJson.entries) {
@@ -40,7 +44,7 @@ void main() {
       method.isRegistered = true;
     }
 
-    editorClient = EditorClient(mockDtd);
+    editorClient = EditorClient(mockDTDManager);
   });
 
   group('getRefactors', () {

packages/devtools_app_shared/lib/src/service/dtd_manager.dart

@@ -32,6 +32,19 @@ class DTDManager {
   Uri? get uri => _uri;
   Uri? _uri;
 
+  /// A stream of [CoreDtdServiceConstants.serviceRegisteredKind] and
+  /// [CoreDtdServiceConstants.serviceUnregisteredKind] events.
+  ///
+  /// Since this is a broadcast stream, it supports multiple subscribers.
+  Stream<DTDEvent> get serviceRegistrationBroadcastStream =>
+      _serviceRegistrationController.stream;
+  final _serviceRegistrationController = StreamController<DTDEvent>.broadcast();
+
+  /// The subscription to the current service registration stream.
+  ///
+  /// This is canceled and reset with the DTD connection changes.
+  StreamSubscription<DTDEvent>? _currentServiceRegistrationSubscription;
+
   /// Whether or not to automatically reconnect if disconnected.
   ///
   /// This will happen by default as long as the disconnect wasn't
@@ -74,7 +87,7 @@ class DTDManager {
   }
 
   /// Triggers a reconnect to the last connected URI if the current state is
-  /// [ConnectionFailedDTDState] (and there was a pervious connection).
+  /// [ConnectionFailedDTDState] (and there was a previous connection).
   Future<void> reconnect() {
     final reconnectFunc = _lastConnectFunc;
     if (_connectionState.value is! ConnectionFailedDTDState ||
@@ -149,11 +162,19 @@ class DTDManager {
 
     try {
       final connection = await _connectWithRetries(uri, maxRetries: maxRetries);
+      await _listenForServiceRegistrationEvents(connection);
 
+      // Save the previous connection so that we can close it after the new
+      // connection is reestablished.
+      final previousConnection = _connection.value;
       _uri = uri;
       // Set this after setting the value of [_uri] so that [_uri] can be used
       // by any listeners of the [_connection] notifier.
       _connection.value = connection;
+      // Close the previous connection.
+      if (previousConnection != null) {
+        await previousConnection.close();
+      }
       _connectionState.value = ConnectedDTDState();
       _log.info('Successfully connected to DTD at: $uri');
 
@@ -230,16 +251,17 @@ class DTDManager {
       // an explicit disconnect.
       _automaticallyReconnect = false;
 
-      // We only clear the connection if we are explicitly disconnecting. In the
-      // case where the connection just dropped, we leave it so that we can
-      // continue to render a page (usually with an overlay).
+      // We only close and clear the connection if we are explicitly
+      // disconnecting. In the case where the connection just dropped, we leave
+      // it so that we can continue to render a page (usually with an overlay).
+      // We only close it once the new connection is established. 
+      if (_connection.value case final connection?) {
+        await connection.close();
+      }
       _connection.value = null;
     }
 
     _periodicConnectionCheck?.cancel();
-    if (_connection.value case final connection?) {
-      await connection.close();
-    }
 
     _connectionState.value = NotConnectedDTDState();
     _uri = null;
@@ -249,9 +271,47 @@ class DTDManager {
 
   Future<void> dispose() async {
     await disconnect();
+    await _currentServiceRegistrationSubscription?.cancel();
+    await _serviceRegistrationController.close();
     _connection.dispose();
   }
 
+  /// Listens for service registration events on the [dtd] connection.
+  Future<void> _listenForServiceRegistrationEvents(
+      DartToolingDaemon dtd) async {
+    // Note: We immediately begin listening for service registration events on
+    // on the new DTD connection before canceling the previous subscription.
+    // This guarantees that we don't miss any events across reconnects.
+    // ignore: cancel_subscriptions, false positive, it is canceled below.
+    final nextServiceRegistrationSubscription = dtd
+        .onEvent(CoreDtdServiceConstants.servicesStreamId)
+        .listen(_forwardServiceRegistrationEvents,
+            onError: _logServiceStreamError);
+    await dtd.streamListen(CoreDtdServiceConstants.servicesStreamId);
+
+    // Cancel the previous subscription.
+    await _currentServiceRegistrationSubscription?.cancel();
+    _currentServiceRegistrationSubscription =
+        nextServiceRegistrationSubscription;
+  }
+
+  /// Forwards service registration events to the
+  /// [_serviceRegistrationController].
+  void _forwardServiceRegistrationEvents(DTDEvent event) {
+    final kind = event.kind;
+    final isRegistrationEvent =
+        kind == CoreDtdServiceConstants.serviceRegisteredKind ||
+            kind == CoreDtdServiceConstants.serviceUnregisteredKind;
+
+    if (isRegistrationEvent) {
+      _serviceRegistrationController.add(event);
+    }
+  }
+
+  void _logServiceStreamError(Object error) {
+    _log.warning('Error in DTD service stream', error);
+  }
+
   /// Returns the workspace roots for the Dart Tooling Daemon connection.
   ///
   /// These roots are set by the tool that started DTD, which may be the IDE,