[Message scheduler] Implement handling one message at a time.

Change-Id: I2294a7f209414a499fce05c0180358ec271349fe
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/386581
Commit-Queue: Keerti Parthasarathy <[email protected]>
Reviewed-by: Brian Wilkerson <[email protected]>

pkg/analysis_server/lib/src/analysis_server.dart

@@ -282,14 +282,17 @@ abstract class AnalysisServer {
     DartFixPromptManager? dartFixPromptManager,
     this.providedByteStore,
     PluginManager? pluginManager,
+    bool retainDataForTesting = false,
   }) : resourceProvider = OverlayResourceProvider(baseResourceProvider),
        pubApi = PubApi(
          instrumentationService,
          httpClient,
          Platform.environment['PUB_HOSTED_URL'],
        ),
        producerGeneratorsForLintRules = AssistProcessor.computeLintRuleMap(),
-       messageScheduler = MessageScheduler() {
+       messageScheduler = MessageScheduler(
+            testView:
+                retainDataForTesting ? MessageSchedulerTestView() : null) {
     messageScheduler.setServer(this);
     // Set the default URI converter. This uses the resource providers path
     // context (unlike the initialized value) which allows tests to override it.

pkg/analysis_server/lib/src/legacy_analysis_server.dart

@@ -387,6 +387,7 @@ class LegacyAnalysisServer extends AnalysisServer {
     DartFixPromptManager? dartFixPromptManager,
     super.providedByteStore,
     super.pluginManager,
+    bool retainDataForTesting = false,
   }) : lspClientConfiguration = lsp.LspClientConfiguration(
          baseResourceProvider.pathContext,
        ),
@@ -404,6 +405,7 @@ class LegacyAnalysisServer extends AnalysisServer {
          requestStatistics: requestStatistics,
          enableBlazeWatcher: enableBlazeWatcher,
          dartFixPromptManager: dartFixPromptManager,
+         retainDataForTesting: retainDataForTesting,
        ) {
     var contextManagerCallbacks = ServerContextManagerCallbacks(
       this,
@@ -548,8 +550,13 @@ class LegacyAnalysisServer extends AnalysisServer {
     sendStatusNotificationNew(status);
   }
 
-  /// Handle a [request] that was read from the communication channel.
-  void handleRequest(Request request, CancelableToken? cancellationToken) {
+  /// Handle a [request] that was read from the communication channel. The completer
+  /// is used to indicate when the request handling is done.
+  void handleRequest(
+    Request request,
+    Completer<void> completer,
+    CancelableToken? cancellationToken,
+  ) {
     var startTime = DateTime.now();
     performance.logRequestTiming(request.clientRequestTime);
 
@@ -594,6 +601,7 @@ class LegacyAnalysisServer extends AnalysisServer {
                 ServerRecentPerformance.slowRequestsThreshold) {
           recentPerformance.slowRequests.add(requestPerformance!);
         }
+        completer.complete();
       },
       (exception, stackTrace) {
         if (exception is InconsistentAnalysisException) {
@@ -620,6 +628,7 @@ class LegacyAnalysisServer extends AnalysisServer {
           var response = Response(request.id, error: error);
           sendResponse(response);
         }
+        completer.complete();
       },
     );
   }
@@ -635,7 +644,6 @@ class LegacyAnalysisServer extends AnalysisServer {
           cancellationToken: cancellationToken,
         ),
       );
-      messageScheduler.notify();
     } else if (requestOrResponse is Response) {
       handleResponse(requestOrResponse);
     }

pkg/analysis_server/lib/src/lsp/handlers/handler_rename.dart

@@ -219,6 +219,11 @@ class RenameHandler extends LspMessageHandler<RenameParams, WorkspaceEdit?> {
           return error(ServerErrorCodes.RenameNotValid, finalStatus.message!);
         }
 
+        // Set the completer to complete to show that request is paused, and
+        // that processing of incoming messages can continue while we wait
+        // for the user's response. 
+        message.completer?.complete();
+
         // Otherwise, ask the user whether to proceed with the rename.
         var userChoice = await prompt(
           MessageType.warning,

pkg/analysis_server/lib/src/lsp/handlers/handlers.dart

@@ -440,6 +440,11 @@ class MessageInfo {
   /// May be null for messages sent prior to initialization.
   final LspClientCapabilities? clientCapabilities;
 
+  /// The completer used to indicate that the handler is paused waiting on user
+  /// response. The completer, if any, is set to complete before a dialog is 
+  /// shown. 
+  final Completer<void>? completer;
+
   MessageInfo({
     required this.performance,
     // TODO(dantup): Consider a version of this that has a non-nullable
@@ -448,6 +453,7 @@ class MessageInfo {
     //  to run without, so it would remove a bunch of boilerplate in the others.
     required this.clientCapabilities,
     this.timeSinceRequest,
+    this.completer,
   });
 }
 
@@ -496,6 +502,9 @@ abstract class ServerStateMessageHandler {
     // will need to specifically check the token after `await`s.
     cancellationToken ??= cancelHandler.createToken(message);
     try {
+      if (cancellationToken.isCancellationRequested) {
+        return cancelled(cancellationToken);
+      }
       var result = await handler.handleMessage(
         message,
         messageInfo,

pkg/analysis_server/lib/src/lsp/lsp_analysis_server.dart

@@ -163,6 +163,7 @@ class LspAnalysisServer extends AnalysisServer {
     // Disable to avoid using this in unit tests.
     bool enableBlazeWatcher = false,
     DartFixPromptManager? dartFixPromptManager,
+    bool retainDataForTesting = false,
   }) : lspClientConfiguration = LspClientConfiguration(
          baseResourceProvider.pathContext,
        ),
@@ -179,6 +180,7 @@ class LspAnalysisServer extends AnalysisServer {
          LspNotificationManager(baseResourceProvider.pathContext),
          enableBlazeWatcher: enableBlazeWatcher,
          dartFixPromptManager: dartFixPromptManager,
+         retainDataForTesting: retainDataForTesting,
        ) {
     notificationManager.server = this;
     messageHandler = UninitializedStateMessageHandler(this);
@@ -495,8 +497,15 @@ class LspAnalysisServer extends AnalysisServer {
     );
   }
 
-  /// Handle a [message] that was read from the communication channel.
-  void handleMessage(Message message, {CancelableToken? cancellationToken}) {
+  /// Handle a [message] that was read from the communication channel. The
+  /// [completer] is used to indicate when the [RequestMessage] handling is
+  /// done. [ResponseMessage] and [NotificationMessage]'s are not waited on
+  /// till finish.
+  void handleMessage(
+    Message message,
+    Completer<void>? completer, {
+    CancelableToken? cancellationToken,
+  }) {
     var startTime = DateTime.now();
     performance.logRequestTiming(message.clientRequestTime);
     runZonedGuarded(() async {
@@ -520,6 +529,7 @@ class LspAnalysisServer extends AnalysisServer {
               performance: performance,
               clientCapabilities: editorClientCapabilities,
               timeSinceRequest: message.timeSinceRequest,
+              completer: completer,
             );
 
             if (message is RequestMessage) {
@@ -551,6 +561,9 @@ class LspAnalysisServer extends AnalysisServer {
         } else {
           showErrorMessageToUser('Unknown message type');
         }
+        if (completer != null && !completer.isCompleted) {
+          completer.complete();
+        }
       } on InconsistentAnalysisException {
         sendErrorResponse(
           message,
@@ -559,6 +572,9 @@ class LspAnalysisServer extends AnalysisServer {
             message: 'Document was modified before operation completed',
           ),
         );
+        if (completer != null && !completer.isCompleted) {
+          completer.complete();
+        }
       } catch (error, stackTrace) {
         var errorMessage =
             message is ResponseMessage
@@ -576,6 +592,9 @@ class LspAnalysisServer extends AnalysisServer {
           ),
         );
         logException(errorMessage, error, stackTrace);
+        if (completer != null && !completer.isCompleted) {
+          completer.complete();
+        }
       }
     }, socketError);
   }
@@ -815,7 +834,6 @@ class LspAnalysisServer extends AnalysisServer {
     messageScheduler.add(
       LspMessage(message: message, cancellationToken: cancellationToken),
     );
-    messageScheduler.notify();
   }
 
   void sendErrorResponse(Message message, ResponseError error) {