Update JsonRpcConnection.cs

src/Draco.JsonRpc/JsonRpcConnection.cs

@@ -60,31 +60,48 @@ private sealed class OutgoingRequest<TResponse> : IOutgoingRequest
     // Communication state
     private int lastMessageId = 0;
 
-    // IO channels
-    private readonly Channel<TMessage> incomingMessages = Channel.CreateUnbounded<TMessage>(new()
-    {
-        SingleReader = true,
-        SingleWriter = true,
-    });
-    private readonly Channel<TMessage> outgoingMessages = Channel.CreateUnbounded<TMessage>(new()
-    {
-        SingleReader = true,
-        SingleWriter = false,
-    });
-
     // Pending requests
     private readonly ConcurrentDictionary<object, CancellationTokenSource> pendingIncomingRequests = new();
     private readonly ConcurrentDictionary<int, IOutgoingRequest> pendingOutgoingRequests = new();
 
     // Handlers
     private readonly Dictionary<string, IJsonRpcMethodHandler> methodHandlers = new();
 
+    // Shutdown
+    private readonly CancellationTokenSource shutdownTokenSource = new();
+
     // TODO: Doc
     public void AddHandler(IJsonRpcMethodHandler handler) => this.methodHandlers.Add(handler.Method, handler);
 
-    // TODO: Start listening
-    // TODO: IO loop
-    // TODO: Shutdown
+    // TODO: Doc
+    public Task StartListening() => Task.Run(async () =>
+    {
+        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)
+            {
+                await this.ProcessIncomingNotificationAsync(message);
+            }
+            else
+            {
+                // TODO: What to do?
+            }
+        }
+    });
+
+    public void Shutdown() => this.shutdownTokenSource.Cancel();
 
     protected int NextMessageId() => Interlocked.Increment(ref this.lastMessageId);