|
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
/*============================================================
**
** Class: AsyncStreamReader
**
** Purpose: For reading text from streams using a particular
** encoding in an asynchronous manner used by the process class
**
**
===========================================================*/
using System.Collections.Generic;
using System.IO;
using System.Runtime.ExceptionServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace System.Diagnostics
{
internal sealed class AsyncStreamReader : IDisposable
{
private const int DefaultBufferSize = 1024; // Byte buffer size
private readonly Stream _stream;
private readonly Decoder _decoder;
private readonly byte[] _byteBuffer;
private readonly char[] _charBuffer;
// Delegate to call user function.
private readonly Action<string?> _userCallBack;
private readonly CancellationTokenSource _cts;
private Task? _readToBufferTask;
private readonly Queue<string?> _messageQueue;
private StringBuilder? _sb;
private bool _bLastCarriageReturn;
private bool _cancelOperation;
// Cache the last position scanned in sb when searching for lines.
private int _currentLinePos;
// Creates a new AsyncStreamReader for the given stream. The
// character encoding is set by encoding and the buffer size,
// in number of 16-bit characters, is set by bufferSize.
internal AsyncStreamReader(Stream stream, Action<string?> callback, Encoding encoding)
{
Debug.Assert(stream != null && encoding != null && callback != null, "Invalid arguments!");
Debug.Assert(stream.CanRead, "Stream must be readable!");
_stream = stream;
_userCallBack = callback;
_decoder = encoding.GetDecoder();
_byteBuffer = new byte[DefaultBufferSize];
// This is the maximum number of chars we can get from one iteration in loop inside ReadBuffer.
// Used so ReadBuffer can tell when to copy data into a user's char[] directly, instead of our internal char[].
int maxCharsPerBuffer = encoding.GetMaxCharCount(DefaultBufferSize);
_charBuffer = new char[maxCharsPerBuffer];
_cts = new CancellationTokenSource();
_messageQueue = new Queue<string?>();
}
// User calls BeginRead to start the asynchronous read
internal void BeginReadLine()
{
_cancelOperation = false;
if (_sb == null)
{
_sb = new StringBuilder(DefaultBufferSize);
_readToBufferTask = ReadBufferAsync();
}
else
{
FlushMessageQueue(rethrowInNewThread: false);
}
}
internal void CancelOperation()
{
_cancelOperation = true;
}
// This is the async callback function. Only one thread could/should call this.
private async Task ReadBufferAsync()
{
await Task.CompletedTask.ConfigureAwait(ConfigureAwaitOptions.ForceYielding);
while (true)
{
try
{
int bytesRead = await _stream.ReadAsync(new Memory<byte>(_byteBuffer), _cts.Token).ConfigureAwait(false);
if (bytesRead == 0)
break;
int charLen = _decoder.GetChars(_byteBuffer, 0, bytesRead, _charBuffer, 0);
_sb!.Append(_charBuffer, 0, charLen);
MoveLinesFromStringBuilderToMessageQueue();
}
catch (IOException)
{
// We should ideally consume errors from operations getting cancelled
// so that we don't crash the unsuspecting parent with an unhandled exc.
// This seems to come in 2 forms of exceptions (depending on platform and scenario),
// namely OperationCanceledException and IOException (for errorcode that we don't
// map explicitly).
break; // Treat this as EOF
}
catch (OperationCanceledException)
{
// We should consume any OperationCanceledException from child read here
// so that we don't crash the parent with an unhandled exc
break; // Treat this as EOF
}
// If user's delegate throws exception we treat this as EOF and
// completing without processing current buffer content
if (FlushMessageQueue(rethrowInNewThread: true))
{
return;
}
}
// We're at EOF, process current buffer content and flush message queue.
lock (_messageQueue)
{
if (_sb!.Length != 0)
{
_messageQueue.Enqueue(_sb.ToString());
_sb.Length = 0;
}
_messageQueue.Enqueue(null);
}
FlushMessageQueue(rethrowInNewThread: true);
}
// Read lines stored in StringBuilder and the buffer we just read into.
// A line is defined as a sequence of characters followed by
// a carriage return ('\r'), a line feed ('\n'), or a carriage return
// immediately followed by a line feed. The resulting string does not
// contain the terminating carriage return and/or line feed. The returned
// value is null if the end of the input stream has been reached.
private void MoveLinesFromStringBuilderToMessageQueue()
{
int currentIndex = _currentLinePos;
int lineStart = 0;
int len = _sb!.Length;
// skip a beginning '\n' character of new block if last block ended
// with '\r'
if (_bLastCarriageReturn && (len > 0) && _sb[0] == '\n')
{
currentIndex = 1;
lineStart = 1;
_bLastCarriageReturn = false;
}
while (currentIndex < len)
{
char ch = _sb[currentIndex];
// Note the following common line feed chars:
// \n - UNIX \r\n - DOS \r - Mac
if (ch == '\r' || ch == '\n')
{
string line = _sb.ToString(lineStart, currentIndex - lineStart);
lineStart = currentIndex + 1;
// skip the "\n" character following "\r" character
if ((ch == '\r') && (lineStart < len) && (_sb[lineStart] == '\n'))
{
lineStart++;
currentIndex++;
}
lock (_messageQueue)
{
_messageQueue.Enqueue(line);
}
}
currentIndex++;
}
if ((len > 0) && _sb[len - 1] == '\r')
{
_bLastCarriageReturn = true;
}
// Keep the rest characters which can't form a new line in string builder.
if (lineStart < len)
{
if (lineStart == 0)
{
// we found no breaklines, in this case we cache the position
// so next time we don't have to restart from the beginning
_currentLinePos = currentIndex;
}
else
{
_sb.Remove(0, lineStart);
_currentLinePos = 0;
}
}
else
{
_sb.Length = 0;
_currentLinePos = 0;
}
}
// If everything runs without exception, returns false.
// If an exception occurs and rethrowInNewThread is true, returns true.
// If an exception occurs and rethrowInNewThread is false, the exception propagates.
private bool FlushMessageQueue(bool rethrowInNewThread)
{
try
{
// Keep going until we're out of data to process.
while (true)
{
// Get the next line (if there isn't one, we're done) and
// invoke the user's callback with it.
string? line;
lock (_messageQueue)
{
if (_messageQueue.Count == 0)
{
break;
}
line = _messageQueue.Dequeue();
}
if (!_cancelOperation)
{
_userCallBack(line); // invoked outside of the lock
}
}
return false;
}
catch (Exception e)
{
// If rethrowInNewThread is true, we can't let the exception propagate synchronously on this thread,
// so propagate it in a thread pool thread and return true to indicate to the caller that this failed.
// Otherwise, let the exception propagate.
if (rethrowInNewThread)
{
ThreadPool.QueueUserWorkItem(edi => ((ExceptionDispatchInfo)edi!).Throw(), ExceptionDispatchInfo.Capture(e));
return true;
}
throw;
}
}
internal Task EOF => _readToBufferTask ?? Task.CompletedTask;
public void Dispose()
{
_cts.Cancel();
}
}
}
|