Browse Source

Merge branch 'dev' of https://github.com/ionite34/StabilityMatrix into invoke-fixes

pull/240/head
JT 1 year ago
parent
commit
407e876bf4
  1. 28
      StabilityMatrix.Core/Processes/AnsiProcess.cs
  2. 65
      StabilityMatrix.Core/Processes/AsyncStreamReader.cs

28
StabilityMatrix.Core/Processes/AnsiProcess.cs

@ -23,9 +23,9 @@ public class AnsiProcess : Process
StartInfo.RedirectStandardError = true;
// Need this to parse ANSI escape sequences correctly
StartInfo.StandardOutputEncoding = Encoding.UTF8;
StartInfo.StandardErrorEncoding = Encoding.UTF8;
StartInfo.StandardInputEncoding = Encoding.UTF8;
StartInfo.StandardOutputEncoding = new UTF8Encoding(false);
StartInfo.StandardErrorEncoding = new UTF8Encoding(false);
StartInfo.StandardInputEncoding = new UTF8Encoding(false);
}
/// <summary>
@ -35,18 +35,28 @@ public class AnsiProcess : Process
public void BeginAnsiRead(Action<ProcessOutput> callback)
{
var stdoutStream = StandardOutput.BaseStream;
stdoutReader = new AsyncStreamReader(stdoutStream, s =>
stdoutReader = new AsyncStreamReader(
stdoutStream,
s =>
{
if (s == null) return;
if (s == null)
return;
callback(ProcessOutput.FromStdOutLine(s));
}, StandardOutput.CurrentEncoding);
},
StandardOutput.CurrentEncoding
);
var stderrStream = StandardError.BaseStream;
stderrReader = new AsyncStreamReader(stderrStream, s =>
stderrReader = new AsyncStreamReader(
stderrStream,
s =>
{
if (s == null) return;
if (s == null)
return;
callback(ProcessOutput.FromStdErrLine(s));
}, StandardError.CurrentEncoding);
},
StandardError.CurrentEncoding
);
stdoutReader.BeginReadLine();
stderrReader.BeginReadLine();

65
StabilityMatrix.Core/Processes/AsyncStreamReader.cs

@ -96,14 +96,18 @@ internal sealed class AsyncStreamReader : IDisposable
{
try
{
var bytesRead = await _stream.ReadAsync(new Memory<byte>(_byteBuffer), _cts.Token).ConfigureAwait(false);
var bytesRead = await _stream
.ReadAsync(new Memory<byte>(_byteBuffer), _cts.Token)
.ConfigureAwait(false);
if (bytesRead == 0)
break;
var charLen = _decoder.GetChars(_byteBuffer, 0, bytesRead, _charBuffer, 0);
Debug.WriteLine($"AsyncStreamReader - Read {charLen} chars: " +
$"{new string(_charBuffer, 0, charLen).ToRepr()}");
Debug.WriteLine(
$"AsyncStreamReader - Read {charLen} chars: "
+ $"{new string(_charBuffer, 0, charLen).ToRepr()}"
);
_sb!.Append(_charBuffer, 0, charLen);
MoveLinesFromStringBuilderToMessageQueue();
@ -140,7 +144,9 @@ internal sealed class AsyncStreamReader : IDisposable
var remaining = _sb.ToString();
_messageQueue.Enqueue(remaining);
_sb.Length = 0;
Debug.WriteLine($"AsyncStreamReader - Reached EOF, sent remaining buffer: {remaining}");
Debug.WriteLine(
$"AsyncStreamReader - Reached EOF, sent remaining buffer: {remaining}"
);
}
_messageQueue.Enqueue(null);
}
@ -154,7 +160,8 @@ internal sealed class AsyncStreamReader : IDisposable
{
lock (_messageQueue)
{
if (_sb!.Length == 0) return;
if (_sb!.Length == 0)
return;
_messageQueue.Enqueue(_sb.ToString());
_sb.Length = 0;
@ -167,7 +174,8 @@ internal sealed class AsyncStreamReader : IDisposable
{
lock (_messageQueue)
{
if (_sb!.Length == 0) return;
if (_sb!.Length == 0)
return;
_messageQueue.Enqueue(_sb.ToString(startIndex, _sb.Length - startIndex));
_sb.Length = 0;
@ -177,7 +185,8 @@ internal sealed class AsyncStreamReader : IDisposable
// Sends a message to the queue if not null or empty
private void SendToQueue(string? message)
{
if (string.IsNullOrEmpty(message)) return;
if (string.IsNullOrEmpty(message))
return;
lock (_messageQueue)
{
_messageQueue.Enqueue(message);
@ -196,14 +205,6 @@ internal sealed class AsyncStreamReader : IDisposable
var lineStart = 0;
var len = _sb!.Length;
// If flagged, send next buffer immediately
if (_sendNextBufferImmediately)
{
SendRemainingBuffer();
_sendNextBufferImmediately = false;
return;
}
// skip a beginning '\n' character of new block if last block ended
// with '\r'
if (_bLastCarriageReturn && len > 0 && _sb[0] == '\n')
@ -255,7 +256,8 @@ internal sealed class AsyncStreamReader : IDisposable
{
// Send buffer up to this point, not including \r
// But skip if there's no content
if (currentIndex == lineStart) break;
if (currentIndex == lineStart)
break;
var line = _sb.ToString(lineStart, currentIndex - lineStart);
lock (_messageQueue)
@ -294,9 +296,13 @@ internal sealed class AsyncStreamReader : IDisposable
// lineStart = searchIndex + 1;
currentIndex = searchIndex;
var remainingStart = currentIndex + 1;
var remainingStr =
_sb.ToString(remainingStart, _sb.Length - remainingStart);
Debug.WriteLine($"AsyncStreamReader - Sending remaining buffer: '{remainingStr}'");
var remainingStr = _sb.ToString(
remainingStart,
_sb.Length - remainingStart
);
Debug.WriteLine(
$"AsyncStreamReader - Sending remaining buffer: '{remainingStr}'"
);
// Send the rest of the buffer immediately
SendRemainingBuffer(currentIndex + 1);
return;
@ -324,7 +330,9 @@ internal sealed class AsyncStreamReader : IDisposable
var escapeSequence = result.Value;
SendToQueue(escapeSequence);
Debug.WriteLine($"AsyncStreamReader - Sent Ansi escape sequence: {escapeSequence.ToRepr()}");
Debug.WriteLine(
$"AsyncStreamReader - Sent Ansi escape sequence: {escapeSequence.ToRepr()}"
);
// Advance currentIndex and lineStart to end of escape sequence
// minus 1 since we will increment currentIndex at the end of the loop
@ -333,7 +341,9 @@ internal sealed class AsyncStreamReader : IDisposable
}
else
{
Debug.WriteLine($"AsyncStreamReader - No match for Ansi escape sequence: {remaining.ToRepr()}");
Debug.WriteLine(
$"AsyncStreamReader - No match for Ansi escape sequence: {remaining.ToRepr()}"
);
}
break;
@ -346,6 +356,14 @@ internal sealed class AsyncStreamReader : IDisposable
_bLastCarriageReturn = true;
}
// If flagged, send remaining buffer immediately
if (_sendNextBufferImmediately)
{
SendRemainingBuffer();
_sendNextBufferImmediately = false;
return;
}
// Keep the rest characters which can't form a new line in string builder.
if (lineStart < len)
{
@ -404,7 +422,10 @@ internal sealed class AsyncStreamReader : IDisposable
// Otherwise, let the exception propagate.
if (rethrowInNewThread)
{
ThreadPool.QueueUserWorkItem(edi => ((ExceptionDispatchInfo)edi!).Throw(), ExceptionDispatchInfo.Capture(e));
ThreadPool.QueueUserWorkItem(
edi => ((ExceptionDispatchInfo)edi!).Throw(),
ExceptionDispatchInfo.Capture(e)
);
return true;
}
throw;

Loading…
Cancel
Save