Update JsonRpcConnection.cs

src/Draco.JsonRpc/JsonRpcConnection.cs

@@ -42,6 +42,13 @@ private sealed class OutgoingRequest<TResponse> : IOutgoingRequest
         public void Complete(object? result) => this.tcs.SetResult((TResponse?)result);
     }
 
+    private enum ConsumerState
+    {
+        None,
+        Concurrent,
+        Exclusive,
+    }
+
     /// <summary>
     /// The IO transport pipeline to send and receive messages through.
     /// </summary>
@@ -76,36 +83,72 @@ private sealed class OutgoingRequest<TResponse> : IOutgoingRequest
     // TODO: Doc
     public Task StartListening() => Task.Run(async () =>
     {
+        var state = ConsumerState.None;
+        var pendingProcesses = new List<Task>();
+
         while (!this.shutdownTokenSource.IsCancellationRequested)
         {
             var message = await this.ReadMessageAsync();
             if (message is null) break;
 
-            if (message.IsRequest)
-            {
-                var response = await this.ProcessIncomingRequestAsync(message);
-                this.outgoingMessages.Writer.TryWrite(response);
-            }
-            else if (message.IsResponse)
-            {
-                this.ProcessIncomingResponse(message);
-            }
-            else if (message.IsNotification)
+            var isExclusive = this.methodHandlers.TryGetValue(message.Method, out var handler)
+                ? handler.Mutating
+                : false;
+
+            if (isExclusive)
             {
-                await this.ProcessIncomingNotificationAsync(message);
+                if (state == ConsumerState.Concurrent)
+                {
+                    // We are in concurrent state, wait for messages to complete
+                    await Task.WhenAll(pendingProcesses);
+                    pendingProcesses.Clear();
+                }
+                // We are now in exclusive state
+                state = ConsumerState.Exclusive;
+                // Process synchronously
+                await this.ProcessIncomingMessage(message);
             }
             else
             {
-                // TODO: What to do?
+                // If we get here, we are guaranteed to be done with exclusive message processing
+                // we can start concurrent processing
+
+                // Add to the pool
+                pendingProcesses.Add(Task.Run(() => this.ProcessIncomingMessage(message)));
+                state = ConsumerState.Concurrent;
             }
         }
+
+        // Process any remaining concurrent message
+        await Task.WhenAll(pendingProcesses);
     });
 
     public void Shutdown() => this.shutdownTokenSource.Cancel();
 
     protected int NextMessageId() => Interlocked.Increment(ref this.lastMessageId);
 
     #region Message Processing
+    private async Task ProcessIncomingMessage(TMessage message)
+    {
+        if (message.IsRequest)
+        {
+            var response = await this.ProcessIncomingRequestAsync(message);
+            this.outgoingMessages.Writer.TryWrite(response);
+        }
+        else if (message.IsResponse)
+        {
+            this.ProcessIncomingResponse(message);
+        }
+        else if (message.IsNotification)
+        {
+            await this.ProcessIncomingNotificationAsync(message);
+        }
+        else
+        {
+            // TODO: What to do?
+        }
+    }
+
     private async Task<TMessage> ProcessIncomingRequestAsync(TMessage message)
     {
         TMessage Error(object error)