Merge pull request #5 from Lercher/master

Fix for Issue #3 and #4

Author: david-driscoll

src/JsonRpc/InputHandler.cs

@@ -87,7 +87,7 @@ private async void ProcessInputStream()
                     current += await _input.ReadBlockAsync(buffer, current, 1);
                 }
 
-                var headersContent = Encoding.ASCII.GetString(Encoding.ASCII.GetBytes(buffer, 0, current));
+                var headersContent = new string(buffer, 0, current);
                 var headers = headersContent.Split(HeaderKeys, StringSplitOptions.RemoveEmptyEntries);
                 long length = 0;
                 for (var i = 0; i < headers.Length; i += 2)
@@ -104,7 +104,7 @@ private async void ProcessInputStream()
 
                 await _input.ReadBlockAsync(requestBuffer, 0, requestBuffer.Length);
 
-                var payload = Encoding.ASCII.GetString(Encoding.ASCII.GetBytes(requestBuffer));
+                var payload = new string(requestBuffer);
 
                 HandleRequest(payload);
             }

src/JsonRpc/OutputHandler.cs

@@ -9,42 +9,38 @@ namespace JsonRpc
 {
     public class OutputHandler : IOutputHandler
     {
-        private readonly TimeSpan _sleepTime = TimeSpan.FromMilliseconds(50);
         private readonly TextWriter _output;
         private Thread _thread;
-        private readonly ConcurrentQueue<object> _queue;
+        private readonly BlockingCollection<object> _queue;
+        private readonly CancellationTokenSource _cancel;
 
         public OutputHandler(TextWriter output)
         {
             _output = output;
-            _queue = new ConcurrentQueue<object>();
+            _queue = new BlockingCollection<object>();
+            _cancel = new CancellationTokenSource();
             _thread = new Thread(ProcessOutputQueue) {
                 IsBackground = true
             };
         }
 
-        internal OutputHandler(TextWriter output, TimeSpan sleepTime)
-            : this(output)
-        {
-            _sleepTime = sleepTime;
-        }
-
         public void Start()
         {
             _thread.Start();
         }
 
         public void Send(object value)
         {
-            _queue.Enqueue(value);
+            _queue.Add(value);
         }
+
         private void ProcessOutputQueue()
         {
+            var token = _cancel.Token;
             while (true)
             {
                 if (_thread == null) return;
-
-                if (_queue.TryDequeue(out var value))
+                if (_queue.TryTake(out var value, -1, token))
                 {
                     var content = JsonConvert.SerializeObject(value);
 
@@ -56,18 +52,14 @@ private void ProcessOutputQueue()
 
                     _output.Write(sb.ToString());
                 }
-
-                if (_queue.IsEmpty)
-                {
-                    Thread.Sleep(_sleepTime);
-                }
             }
         }
 
         public void Dispose()
         {
             _output?.Dispose();
             _thread = null;
+            _cancel.Cancel();
         }
     }
 }

test/JsonRpc.Tests/OutputHandlerTests.cs

@@ -15,7 +15,7 @@ private static (OutputHandler handler, Action wait) NewHandler(TextWriter textWr
                 cts.CancelAfter(TimeSpan.FromSeconds(5));
             action(cts);
 
-            var handler = new OutputHandler(textWriter, TimeSpan.Zero);
+            var handler = new OutputHandler(textWriter);
             handler.Start();
             return (handler, () => cts.Wait());
         }