Refactored the process scheduling component of Input handler into it's own class.

Author: Martin Lercher

src/JsonRpc/InputHandler.cs

@@ -1,12 +1,8 @@
 using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.IO;
 using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using JsonRpc.Server;
 using JsonRpc.Server.Messages;
 using Newtonsoft.Json.Linq;
 
@@ -27,9 +23,7 @@ public class InputHandler : IInputHandler
         private Thread _inputThread;
         private readonly IRequestRouter _requestRouter;
         private readonly IResponseRouter _responseRouter;
-        private readonly BlockingCollection<(RequestProcessType type, Func<Task> request)> _queue;
-        private readonly CancellationTokenSource _cancelQueue;
-        private Thread _queueThread;
+        private readonly ProcessScheduler _scheduler;
 
         public InputHandler(
             TextReader input,
@@ -46,19 +40,16 @@ IResponseRouter responseRouter
             _requestProcessIdentifier = requestProcessIdentifier;
             _requestRouter = requestRouter;
             _responseRouter = responseRouter;
-            _queue = new BlockingCollection<(RequestProcessType type, Func<Task> request)>();
-            _cancelQueue = new CancellationTokenSource();
 
+            _scheduler = new ProcessScheduler();
             _inputThread = new Thread(ProcessInputStream) { IsBackground = true };
-
-            _queueThread = new Thread(ProcessRequestQueue) { IsBackground = true };
         }
 
         public void Start()
         {
             _outputHandler.Start();
             _inputThread.Start();
-            _queueThread.Start();
+            _scheduler.Start();
         }
 
         private async void ProcessInputStream()
@@ -145,24 +136,24 @@ private void HandleRequest(string request)
             {
                 if (item.IsRequest)
                 {
-                    _queue.Add((
+                    _scheduler.Add(
                         type,
                         async () => {
                             var result = await _requestRouter.RouteRequest(item.Request);
 
                             _outputHandler.Send(result.Value);
                         }
-                    ));
+                    );
                 }
                 else if (item.IsNotification)
                 {
-                    _queue.Add((
+                    _scheduler.Add(
                         type,
                         () => {
                             _requestRouter.RouteNotification(item.Notification);
                             return Task.CompletedTask;
                         }
-                    ));
+                    );
                 }
                 else if (item.IsError)
                 {
@@ -172,46 +163,12 @@ private void HandleRequest(string request)
             }
         }
 
-        private Task Start(Func<Task> request)
-        {
-            var t = request();
-            t.Start();
-            return t;
-        }
-
-        private async void ProcessRequestQueue()
-        {
-            // see https://github.com/OmniSharp/csharp-language-server-protocol/issues/4
-            var token = _cancelQueue.Token;
-            var waitables = new List<Task>();
-            while(true)
-            {
-                if (_queueThread == null) return;
-                if (_queue.TryTake(out var item, Timeout.Infinite, token))
-                {
-                    var (type, request) = item;
-                    if (type == RequestProcessType.Serial)
-                    {
-                        await Task.WhenAll(waitables);
-                        waitables.Clear();
-                        await Start(request);
-                    }
-                    else if (type == RequestProcessType.Parallel)
-                    {
-                        waitables.Add(Start(request));
-                    }
-                    else
-                        throw new NotImplementedException("Only Serial and Parallel execution types can be handled currently");
-                }
-            }
-        }
 
         public void Dispose()
         {
             _outputHandler.Dispose();
             _inputThread = null;
-            _queueThread = null;
-            _cancelQueue.Cancel();
+            _scheduler?.Dispose();
         }
     }
 }

src/JsonRpc/ProcessScheduler.cs

@@ -0,0 +1,70 @@
+using System;
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JsonRpc
+{
+    public class ProcessScheduler : IDisposable
+    {
+        private readonly BlockingCollection<(RequestProcessType type, Func<Task> request)> _queue;
+        private readonly CancellationTokenSource _cancel;
+        private Thread _queueThread;
+
+        public ProcessScheduler() {
+            _queue = new BlockingCollection<(RequestProcessType type, Func<Task> request)>();
+            _cancel = new CancellationTokenSource();
+            _queueThread = new Thread(ProcessRequestQueue) { IsBackground = true };
+        }
+
+        public void Start() {
+            _queueThread.Start();
+        }
+
+        public void Add(RequestProcessType type, Func<Task> request) {
+            _queue.Add((type, request));
+        }
+
+        private Task Start(Func<Task> request)
+        {
+            var t = request();
+            t.Start();
+            return t;
+        }
+
+        private void ProcessRequestQueue()
+        {
+            // see https://github.com/OmniSharp/csharp-language-server-protocol/issues/4
+            // no need to be async, because this thing already allocated a thread on it's own.
+            var token = _cancel.Token;
+            var waitables = new List<Task>();
+            while(true)
+            {
+                if (_queueThread == null) return;
+                if (_queue.TryTake(out var item, Timeout.Infinite, token))
+                {
+                    var (type, request) = item;
+                    if (type == RequestProcessType.Serial)
+                    {
+                        Task.WaitAll(waitables.ToArray(), token);
+                        waitables.Clear();
+                        Start(request).Wait(token);
+                    }
+                    else if (type == RequestProcessType.Parallel)
+                    {
+                        waitables.Add(Start(request));
+                    }
+                    else
+                        throw new NotImplementedException("Only Serial and Parallel execution types can be handled currently");
+                }
+            }
+        }
+
+        public void Dispose()
+        {
+            _queueThread = null;
+            _cancel.Cancel();
+        }
+    }
+}