Skip to content

JsonRpcServer now disposable #168

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Sep 14, 2019
Merged

JsonRpcServer now disposable #168

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4908f44
JsonRpcServer.Dispose() works now
TylerLeonhardt Sep 11, 2019
4605713
additional changes for disposable
TylerLeonhardt Sep 11, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 49 additions & 41 deletions src/JsonRpc/InputHandler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,52 +72,59 @@ private void ProcessInputStream()
// content is encoded in UTF-8
while (true)
{
if (_inputThread == null) return;

var buffer = new byte[300];
var current = _input.Read(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)
{
var n = _input.Read(buffer, current, 1);
if (n == 0) return; // no more _input, mitigates endless loop here.
current += n;
}
try {
if (_inputThread == null) return;

var buffer = new byte[300];
var current = _input.Read(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)
{
var n = _input.Read(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 = 1; i < headers.Length; i += 2)
{
// 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))
var headersContent = System.Text.Encoding.ASCII.GetString(buffer, 0, current);
var headers = headersContent.Split(HeaderKeys, StringSplitOptions.RemoveEmptyEntries);
long length = 0;
for (var i = 1; i < headers.Length; i += 2)
{
length = 0;
long.TryParse(value, out length);
// 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 = 0;
long.TryParse(value, out length);
}
}
}

if (length == 0 || length >= int.MaxValue)
{
HandleRequest(string.Empty);
}
else
{
var requestBuffer = new byte[length];
var received = 0;
while (received < length)
if (length == 0 || length >= int.MaxValue)
{
HandleRequest(string.Empty);
}
else
{
var n = _input.Read(requestBuffer, received, requestBuffer.Length - received);
if (n == 0) return; // no more _input
received += n;
var requestBuffer = new byte[length];
var received = 0;
while (received < length)
{
var n = _input.Read(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);
}
// 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);
}
catch (IOException)
{
_logger.LogError("Input stream has been closed.");
break;
}
}
}
Expand Down Expand Up @@ -229,9 +236,10 @@ private void HandleRequest(string request)

public void Dispose()
{
_scheduler.Dispose();
_outputHandler.Dispose();
_inputThread = null;
_scheduler.Dispose();
_input?.Dispose();
}
}
}
12 changes: 8 additions & 4 deletions src/JsonRpc/OutputHandler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ public OutputHandler(Stream output, ISerializer serializer)
_serializer = serializer;
_queue = new BlockingCollection<object>();
_cancel = new CancellationTokenSource();
_outputIsFinished = new TaskCompletionSource<object>();
_thread = new Thread(ProcessOutputQueue) { IsBackground = true, Name = "ProcessOutputQueue" };
}

Expand All @@ -41,7 +42,7 @@ private void ProcessOutputQueue()
var token = _cancel.Token;
try
{
while (true)
while (!token.IsCancellationRequested)
{
if (_queue.TryTake(out var value, Timeout.Infinite, token))
{
Expand All @@ -59,20 +60,23 @@ private void ProcessOutputQueue()
{
ms.Write(headerBytes, 0, headerBytes.Length);
ms.Write(contentBytes, 0, contentBytes.Length);
_output.Write(ms.ToArray(), 0, (int)ms.Position);
if(!token.IsCancellationRequested)
{
_output.Write(ms.ToArray(), 0, (int)ms.Position);
}
}
}
}
}
catch (OperationCanceledException ex)
{
if (ex.CancellationToken != token)
_outputIsFinished.SetException(ex);
_outputIsFinished.TrySetException(ex);
// else ignore. Exceptions: OperationCanceledException - The CancellationToken has been canceled.
}
catch (Exception e)
{
_outputIsFinished.SetException(e);
_outputIsFinished.TrySetException(e);
}
}

Expand Down
2 changes: 1 addition & 1 deletion src/JsonRpc/ProcessScheduler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ private void ProcessRequestQueue()
var waitables = new List<Task>();
try
{
while (true)
while (!token.IsCancellationRequested)
{
if (_queue.TryTake(out var item, Timeout.Infinite, token))
{
Expand Down