Ensure that the scheduler flushes the event queue

There was only one place that wasn't using a future with a zero length
duration, but that's now been fixed. I haven't been able to observe any
performance improvement from this change, but it does have the benefit
that the code is now consistent.

Change-Id: I232728ef2ef9a1000938d0117f611a96c17e9f38
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/435662
Commit-Queue: Brian Wilkerson <brianwilkerson@google.com>
Reviewed-by: Keerti Parthasarathy <keertip@google.com>

pkg/analysis_server/integration_test/server/message_scheduler_test.dart

@@ -341,24 +341,28 @@ Entering process messages loop
   Complete LspMessage: lsp:initialized
 Exit process messages loop
 Incoming RequestMessage: lsp:textDocument/hover
+Incoming RequestMessage: lsp:textDocument/hover
 Entering process messages loop
   Start LspMessage: lsp:textDocument/hover
   Complete LspMessage: lsp:textDocument/hover
-Exit process messages loop
-Incoming RequestMessage: lsp:textDocument/hover
-Entering process messages loop
   Start LspMessage: lsp:textDocument/hover
   Complete LspMessage: lsp:textDocument/hover
 Exit process messages loop
 Pause requested - there are now 1 pauses
 Incoming RequestMessage: lsp:textDocument/hover
 Incoming RequestMessage: lsp:textDocument/hover
+Entering process messages loop
+Exit process messages loop
 Pause requested - there are now 2 pauses
 Incoming RequestMessage: lsp:textDocument/hover
 Incoming RequestMessage: lsp:textDocument/hover
+Entering process messages loop
+Exit process messages loop
 Resume requested - there are now 1 pauses
 Incoming RequestMessage: lsp:textDocument/hover
 Incoming RequestMessage: lsp:textDocument/hover
+Entering process messages loop
+Exit process messages loop
 Resume requested - there are now 0 pauses
 Entering process messages loop
   Start LspMessage: lsp:textDocument/hover

pkg/analysis_server/lib/src/scheduler/message_scheduler.dart

@@ -43,7 +43,7 @@ final class MessageScheduler {
       ListQueue<ScheduledMessage>();
 
   /// Whether the [MessageScheduler] is currently processing messages.
-  bool _isProcessing = false;
+  bool _processingIsScheduled = false;
 
   /// The number of times [pause] has been called without matching [resume]s.
   ///
@@ -180,8 +180,9 @@ final class MessageScheduler {
       }
     }
     _pendingMessages.addLast(message);
-    if (!_isProcessing) {
-      processMessages();
+    if (!_processingIsScheduled) {
+      _processingIsScheduled = true;
+      Future.delayed(Duration.zero, processMessages);
     }
   }
 
@@ -199,14 +200,12 @@ final class MessageScheduler {
 
   /// Dispatch the first message in the queue to be executed.
   void processMessages() async {
-    if (isPaused) {
-      return;
-    }
-
-    _isProcessing = true;
     listener?.startProcessingMessages();
     try {
-      while (_pendingMessages.isNotEmpty && !isPaused) {
+      while (_pendingMessages.isNotEmpty) {
+        if (isPaused) {
+          break;
+        }
         var currentMessage = _pendingMessages.removeFirst();
         _activeMessages.addLast(currentMessage);
         listener?.addActiveMessage(currentMessage);
@@ -269,7 +268,7 @@ final class MessageScheduler {
         server.crashReportingAttachmentsBuilder.forException(error),
       );
     }
-    _isProcessing = false;
+    _processingIsScheduled = false;
     listener?.endProcessingMessages();
   }
 
@@ -280,10 +279,11 @@ final class MessageScheduler {
     }
     _pauseCount--;
     listener?.resumeProcessingMessages(_pauseCount);
-    if (!isPaused && !_isProcessing) {
+    if (!isPaused && !_processingIsScheduled) {
       // Process on the next tick so that the caller to resume() doesn't get
       // messages in the queue attributed to their time (or run before they
       // complete).
+      _processingIsScheduled = true;
       Future.delayed(Duration.zero, processMessages);
     }
   }