source: Common/AsyncStreamReaderFast.cs

using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Diagnostics;

namespace Common
{
    internal delegate void UserCallBack(string data);

    internal class AsyncStreamReaderFast : IDisposable
    {
        internal const int DefaultBufferSize = 16000;  // Byte buffer size
        private const int MinBufferSize = 128;

        private Stream stream;
        private Encoding encoding;
        private Decoder decoder;
        private byte[] byteBuffer;
        private char[] charBuffer;
        // Record the number of valid bytes in the byteBuffer, for a few checks. 

        // This is the maximum number of chars we can get from one call to
        // ReadBuffer.  Used so ReadBuffer can tell when to copy data into
        // a user's char[] directly, instead of our internal char[]. 
        private int maxCharsPerBuffer;

        // Store a backpointer to the process class, to check for user callbacks 
        private StringBuilder sb;

        // Delegate to call user function.
        private UserCallBack userCallBack;

        // Internal Cancel operation 
        private bool cancelOperation;
        private ManualResetEvent eofEvent;

        internal AsyncStreamReaderFast(Process process, Stream stream, UserCallBack callback, Encoding encoding)
            : this(process, stream, callback, encoding, DefaultBufferSize)
        {
        }

        // 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 AsyncStreamReaderFast(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize)
        {
            Debug.Assert(process != null && stream != null && encoding != null && callback != null, "Invalid arguments!");
            Debug.Assert(stream.CanRead, "Stream must be readable!");
            Debug.Assert(bufferSize > 0, "Invalid buffer size!");

            Init(process, stream, callback, encoding, bufferSize);
        }

        private void Init(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize)
        {
            this.stream = stream;
            this.encoding = encoding;
            this.userCallBack = callback;
            decoder = encoding.GetDecoder();
            if (bufferSize < MinBufferSize) bufferSize = MinBufferSize;
            byteBuffer = new byte[bufferSize];
            maxCharsPerBuffer = encoding.GetMaxCharCount(bufferSize);
            charBuffer = new char[maxCharsPerBuffer];
            cancelOperation = false;
            eofEvent = new ManualResetEvent(false);
            sb = new StringBuilder(charBuffer.Length);
        }

        public virtual void Close()
        {
            Dispose(true);
        }

        void IDisposable.Dispose()
        {
            Dispose(true);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (stream != null)
                    stream.Close();
            }
            if (stream != null)
            {
                stream = null;
                encoding = null;
                decoder = null;
                byteBuffer = null;
                charBuffer = null;
            }

            if (eofEvent != null)
            {
                eofEvent.Close();
                eofEvent = null;
            }
        }

        public virtual Encoding CurrentEncoding => encoding;

        public virtual Stream BaseStream => stream;

        // User calls BeginRead to start the asynchronous read
        public void BeginRead()
        {
            if (cancelOperation)
            {
                cancelOperation = false;
            }

            stream.BeginRead(byteBuffer, 0, byteBuffer.Length, ReadBuffer, null);
        }

        internal void CancelOperation()
        {
            cancelOperation = true;
        }

        // This is the async callback function. Only one thread could/should call this.
        private void ReadBuffer(IAsyncResult ar)
        {
            int byteLen;

            try
            {
                byteLen = stream.EndRead(ar);
            }
            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). 
                byteLen = 0; // 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
                byteLen = 0; // Treat this as EOF 
            }

            if (byteLen == 0)
            {
                eofEvent.Set();
            }
            else
            {
                int charLen = decoder.GetChars(byteBuffer, 0, byteLen, charBuffer, 0);
                sb.Length = 0;
                sb.Append(charBuffer, 0, charLen);
                userCallBack(sb.ToString());
                stream.BeginRead(byteBuffer, 0, byteBuffer.Length, ReadBuffer, null);
            }
        }

        // Wait until we hit EOF. This is called from Process.WaitForExit 
        // We will lose some information if we don't do this.
        internal void WaitUtilEOF()
        {
            if (eofEvent == null) return;
            eofEvent.WaitOne();
            eofEvent.Close();
            eofEvent = null;
        }
    }
}