merged omnisharp master. resolved conflicts in 2 files.

Author: Martin Lercher

.appveyor.yml

@@ -9,5 +9,4 @@ after_build:
 artifacts:
   - path: ./artifacts/nuget/*.nupkg
 nuget:
-  account_feed: true
   project_feed: true

nuget.config

@@ -1,7 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <configuration>
   <packageSources>
-    <add key="Build Packages" value="https://www.myget.org/F/c037199d-41df-4567-b966-25ff65324688/api/v3/index.json" />
     <add key="NuGet.org" value="https://api.nuget.org/v3/index.json" />
   </packageSources>
 </configuration>

sample/SampleServer/Program.cs

@@ -29,8 +29,7 @@ static async Task MainAsync(string[] args)
             server.AddHandler(new TextDocumentHandler(server));
 
             await server.Initialize();
-
-            await new TaskCompletionSource<object>().Task;
+            await server.WasShutDown;
         }
     }
 

src/JsonRpc/IScheduler.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Threading.Tasks;
+
+namespace JsonRpc
+{
+    public interface IScheduler : IDisposable
+    {
+        void Start();
+        void Add(RequestProcessType type, Func<Task> request);
+    }
+}

src/JsonRpc/InputHandler.cs

@@ -1,20 +1,15 @@
 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;
 
 namespace JsonRpc
 {
     public class InputHandler : IInputHandler
     {
-        private readonly TimeSpan _sleepTime = TimeSpan.FromMilliseconds(50);
         public const char CR = '\r';
         public const char LF = '\n';
         public static char[] CRLF = { CR, LF };
@@ -28,8 +23,7 @@ public class InputHandler : IInputHandler
         private Thread _inputThread;
         private readonly IRequestRouter _requestRouter;
         private readonly IResponseRouter _responseRouter;
-        private readonly ConcurrentQueue<(RequestProcessType type, Func<Task> request)> _queue;
-        private Thread _queueThread;
+        private readonly IScheduler _scheduler;
 
         public InputHandler(
             Stream input,
@@ -47,31 +41,16 @@ IResponseRouter responseRouter
             _requestProcessIdentifier = requestProcessIdentifier;
             _requestRouter = requestRouter;
             _responseRouter = responseRouter;
-            _queue = new ConcurrentQueue<(RequestProcessType type, Func<Task> request)>();
 
-            _inputThread = new Thread(ProcessInputStream) { IsBackground = true };
-
-            _queueThread = new Thread(ProcessRequestQueue) { IsBackground = true };
-        }
-
-        internal InputHandler(
-            Stream input,
-            IOutputHandler outputHandler,
-            IReciever reciever,
-            IRequestProcessIdentifier requestProcessIdentifier,
-            IRequestRouter requestRouter,
-            IResponseRouter responseRouter,
-            TimeSpan sleepTime
-        ) : this(input, outputHandler, reciever, requestProcessIdentifier, requestRouter, responseRouter)
-        {
-            _sleepTime = sleepTime;
-        }
+            _scheduler = new ProcessScheduler();
+            _inputThread = new Thread(ProcessInputStream) { IsBackground = true, Name = "ProcessInputStream" };
+            }
 
         public void Start()
         {
             _outputHandler.Start();
             _inputThread.Start();
-            _queueThread.Start();
+            _scheduler.Start();
         }
 
         private async void ProcessInputStream()
@@ -85,33 +64,49 @@ private async void ProcessInputStream()
 
                 var buffer = new byte[300];
                 var current = await _input.ReadAsync(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.ReadAsync(buffer, current, 1);
+                    var n = await _input.ReadAsync(buffer, current, 1);
+                    if (n == 0) return; // no more _input, mitigates endless loop here.
+                    current += n;
                 }
 
                 var headersContent = System.Text.Encoding.ASCII.GetString(buffer, 0, current);
                 var headers = headersContent.Split(HeaderKeys, StringSplitOptions.RemoveEmptyEntries);
                 long length = 0;
-                for (var i = 0; i < headers.Length; i += 2)
+                for (var i = 1; i < headers.Length; i += 2)
                 {
-                    var header = headers[0];
-                    var value = headers[i + 1].Trim();
+                    // starting at i = 1 instead of 0 won't throw, if we have uneven headers' length
+                    var header = headers[i - 1];
+                    var value = headers[i].Trim();
                     if (header.Equals("Content-Length", StringComparison.OrdinalIgnoreCase))
                     {
-                        length = long.Parse(value);
+                        length = 0;
+                        long.TryParse(value, out length);
                     }
                 }
 
-                var requestBuffer = new byte[length];
-
-                await _input.ReadAsync(requestBuffer, 0, requestBuffer.Length);
-
-                var payload = System.Text.Encoding.UTF8.GetString(requestBuffer);
-
-                HandleRequest(payload);
+                if (length == 0 || length >= int.MaxValue)
+                {
+                    HandleRequest(string.Empty);
+                }
+                else
+                {
+                    var requestBuffer = new byte[length];
+                    var received = 0;
+                    while (received < length)
+                    {
+                        var n = await _input.ReadAsync(requestBuffer, received, requestBuffer.Length - received);
+                        if (n == 0) return; // no more _input
+                        received += n;
+                    }
+                    // TODO sometimes: encoding should be based on the respective header (including the wrong "utf8" value)
+                    var payload = System.Text.Encoding.UTF8.GetString(requestBuffer); 
+                    HandleRequest(payload);
+                }
             }
         }
 
@@ -162,24 +157,23 @@ private void HandleRequest(string request)
             {
                 if (item.IsRequest)
                 {
-                    _queue.Enqueue((
+                    _scheduler.Add(
                         type,
                         async () => {
                             var result = await _requestRouter.RouteRequest(item.Request);
-
                             _outputHandler.Send(result.Value);
                         }
-                    ));
+                    );
                 }
                 else if (item.IsNotification)
                 {
-                    _queue.Enqueue((
+                    _scheduler.Add(
                         type,
                         () => {
                             _requestRouter.RouteNotification(item.Notification);
                             return Task.CompletedTask;
                         }
-                    ));
+                    );
                 }
                 else if (item.IsError)
                 {
@@ -189,42 +183,12 @@ private void HandleRequest(string request)
             }
         }
 
-        private bool IsNextSerial()
-        {
-            return _queue.TryPeek(out var queueResult) && queueResult.type == RequestProcessType.Serial;
-        }
-
-        private async void ProcessRequestQueue()
-        {
-            while (true)
-            {
-                if (_queueThread == null) return;
-                var items = new List<Func<Task>>();
-                while (!_queue.IsEmpty)
-                {
-                    if (IsNextSerial() && items.Count > 0)
-                    {
-                        break;
-                    }
-
-                    if (_queue.TryDequeue(out var queueResult))
-                        items.Add(queueResult.request);
-                }
-
-                await Task.WhenAll(items.Select(x => x()));
-
-                if (_queue.IsEmpty)
-                {
-                    await Task.Delay(_sleepTime);
-                }
-            }
-        }
 
         public void Dispose()
         {
             _outputHandler.Dispose();
             _inputThread = null;
-            _queueThread = null;
+            _scheduler.Dispose();
         }
     }
 }

src/JsonRpc/OutputHandler.cs

@@ -10,7 +10,7 @@ namespace JsonRpc
     public class OutputHandler : IOutputHandler
     {
         private readonly Stream _output;
-        private Thread _thread;
+        private readonly Thread _thread;
         private readonly BlockingCollection<object> _queue;
         private readonly CancellationTokenSource _cancel;
 
@@ -20,9 +20,7 @@ public OutputHandler(Stream 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()
@@ -38,10 +36,9 @@ public void Send(object value)
         private void ProcessOutputQueue()
         {
             var token = _cancel.Token;
-            while (true)
+            try
             {
-                if (_thread == null) return;
-                try
+                while (true)
                 {
                     if (_queue.TryTake(out var value, Timeout.Infinite, token))
                     {
@@ -63,15 +60,21 @@ private void ProcessOutputQueue()
                         }
                     }
                 }
-                catch (OperationCanceledException) { }
+            }
+            catch (OperationCanceledException ex)
+            {
+                if (ex.CancellationToken != token)
+                    throw;
+                // else ignore. Exceptions: OperationCanceledException - The CancellationToken has been canceled.
             }
         }
 
         public void Dispose()
         {
-            _output?.Dispose();
-            _thread = null;
             _cancel.Cancel();
+            _thread.Join();
+            _cancel.Dispose();
+            _output.Dispose();
         }
     }
 }