Named three background threads

Author: Martin Lercher

src/JsonRpc/InputHandler.cs

@@ -42,8 +42,8 @@ IResponseRouter responseRouter
             _responseRouter = responseRouter;
 
             _scheduler = new ProcessScheduler();
-            _inputThread = new Thread(ProcessInputStream) { IsBackground = true };
-        }
+            _inputThread = new Thread(ProcessInputStream) { IsBackground = true, Name = "ProcessInputStream" };
+            }
 
         public void Start()
         {
@@ -60,10 +60,14 @@ private async void ProcessInputStream()
 
                 var buffer = new char[300];
                 var current = await _input.ReadBlockAsync(buffer, 0, MinBuffer);
+                if (current == 0) return; // no more _input
+
                 while (current < MinBuffer || buffer[current - 4] != CR || buffer[current - 3] != LF ||
                        buffer[current - 2] != CR || buffer[current - 1] != LF)
                 {
-                    current += await _input.ReadBlockAsync(buffer, current, 1);
+                    var n = await _input.ReadBlockAsync(buffer, current, 1);
+                    if (n == 0) return; // no more _input, mitigates endless loop here.
+                    current += n;
                 }
 
                 var headersContent = new string(buffer, 0, current);
@@ -75,17 +79,28 @@ private async void ProcessInputStream()
                     var value = headers[i + 1].Trim();
                     if (header.Equals("Content-Length", StringComparison.OrdinalIgnoreCase))
                     {
-                        length = long.Parse(value);
+                        length = 0;
+                        long.TryParse(value, out length);
                     }
                 }
 
-                var requestBuffer = new char[length];
-
-                await _input.ReadBlockAsync(requestBuffer, 0, requestBuffer.Length);
-
-                var payload = new string(requestBuffer);
-
-                HandleRequest(payload);
+                if (length == 0 || length >= int.MaxValue)
+                {
+                    HandleRequest(string.Empty);
+                }
+                else
+                {
+                    var requestBuffer = new char[length];
+                    var received = 0;
+                    while (received < length)
+                    {
+                        var n = await _input.ReadBlockAsync(requestBuffer, received, requestBuffer.Length - received);
+                        if (n == 0) return; // no more _input
+                        received += n;
+                    }
+                    var payload = new string(requestBuffer);
+                    HandleRequest(payload);
+                }
             }
         }
 
@@ -140,7 +155,6 @@ private void HandleRequest(string request)
                         type,
                         async () => {
                             var result = await _requestRouter.RouteRequest(item.Request);
-
                             _outputHandler.Send(result.Value);
                         }
                     );
@@ -168,7 +182,7 @@ public void Dispose()
         {
             _outputHandler.Dispose();
             _inputThread = null;
-            _scheduler?.Dispose();
+            _scheduler.Dispose();
         }
     }
 }

src/JsonRpc/OutputHandler.cs

@@ -19,9 +19,7 @@ public OutputHandler(TextWriter output)
             _output = output;
             _queue = new BlockingCollection<object>();
             _cancel = new CancellationTokenSource();
-            _thread = new Thread(ProcessOutputQueue) {
-                IsBackground = true
-            };
+            _thread = new Thread(ProcessOutputQueue) { IsBackground = true, Name = "ProcessOutputQueue" };
         }
 
         public void Start()

src/JsonRpc/ProcessScheduler.cs

@@ -10,18 +10,18 @@ public class ProcessScheduler : IScheduler
     {
         private readonly BlockingCollection<(RequestProcessType type, Func<Task> request)> _queue;
         private readonly CancellationTokenSource _cancel;
-        private readonly Thread _queueThread;
+        private readonly Thread _thread;
 
         public ProcessScheduler()
         {
             _queue = new BlockingCollection<(RequestProcessType type, Func<Task> request)>();
             _cancel = new CancellationTokenSource();
-            _queueThread = new Thread(ProcessRequestQueue) { IsBackground = true };
+            _thread = new Thread(ProcessRequestQueue) { IsBackground = true, Name = "ProcessRequestQueue" };
         }
 
         public void Start()
         {
-            _queueThread.Start();
+            _thread.Start();
         }
 
         public void Add(RequestProcessType type, Func<Task> request)
@@ -84,7 +84,7 @@ private void ProcessRequestQueue()
         public void Dispose()
         {
             _cancel.Cancel();
-            _queueThread.Join();
+            _thread.Join();
             _cancel.Dispose();
         }
     }