How to get all available characters from a PipeStream and return immediately?

Question

I'm trying to implement routing the Console output for part of the application into a TextBox. The problem is similar to Elegant Log Window in WinForms C# except that I want to capture all the output from a different stream (the Console output) and show it in the text box instead of explicitly calling a logger method.

My plan was to implement this with a System.IO.Pipes.PipeStream. The Console output would be rerouted as input to the PipeStream. The output from the PipeStream would be read at regular intervals (say once per second) from a timer on the form, and updated to the TextBox, discarding old text to stay under a certain character limit but allowing the user to scroll back say 200 lines.

I initialize the PipeStreams like this:

        private void initPipelines()
        {
            pipeServer = new AnonymousPipeServerStream();
            pipeClient = new AnonymousPipeClientStream(pipeServer.GetClientHandleAsString());
            textWriter_ = new StreamWriter(pipeServer);
            textReader_ = new StreamReader(pipeClient);
            textWriter_.AutoFlush = true;
        }

The routine running once per second to update the text in the TextBox looks like this:

        private void updateText()
        {
            // Get the characters written since the last update
            // I tried this but it always waits:
            // string message = textReader_.ReadToEnd();
            // I tried this but it also waits when there are fewer than 1024 characters available:
            int messageLength = textReader_.ReadBlock(buffer, 0, 1024);
            string message = new string(buffer, 0, messageLength);

            // ... Add the message to the text box and get rid of old text ...
            
        }

In my test routine I simply made a second timer that writes text to the textWriter_ ever 100 to 250 ms.

My problem is that the call to textReader_.ReadToEnd() simply waits, apparently because the textWriter_ stream is still open, and a call to textReader_.ReadBlock also waits if there aren't as many characters to read as will fill the buffer.

What I would like is to have a method for the pipeline that simply reads all the available characters and returns immediately, but I've been unable to find such a function. Is there a way to do this? Or is there some way to determine how many characters would be immediately available?

I need to get the console output character by character, not line-based, since some of the output are "poor man's progress bars" outputting X's to the screen to document progress of the application. All of that is something I can't easily influence.

I thought it would be better to use a System class to accomplish the buffering between the two streams instead of trying to roll my own, but I'm starting to think it might be the only way to do it.

Answer 1

The issue here isn't the pipe object. It's the StreamReader you've wrapped it with.

StreamReader provides four ways consume the decoded text: one character, a buffer of characters, one line, or a complete stream at a time. The latter two are unsuitable for your scenario, but the other two approaches would work.

The ReadBlock() method is a variation on this (specifically, the second technique), but as the name of the method indicates, it blocks until the buffer has been filled or the end of the stream is reached. Instead, you should be using the non-blocking Read() method. This will return whatever characters are currently available, except that it will block long enough for something to be read (i.e. it will only return with a character count of 0 if the end of stream has been reached).

Note that the documentation as written is misleading. While this method will in fact return without filling the provided buffer even if the end of stream hasn't been reached, the current description reads "This method returns after either the number of characters specified by the count parameter are read, or the end of the file is reached." This looks like a copy/paste error, probably which occurred when the ReadBlock() method was introduced in .NET 4.5 and they saw a need to add the clarification.

Answer 2

I resolved the problem by making my own WritableRingBuffer class which derives from TextWriter but also includes a method public string getAllAvailableCharacters() which immediately returns all the characters available, or an empty string if there are none.

Here's the code for it, maybe someone else will find it useful.

using System;
using System.IO;
using System.Text;

namespace CPHUtils
{
    /// <summary>
    /// Class allows writing to a character buffer from which the text can be retrieved
    /// by another object at a later point. It is designed as a ring buffer
    /// with a constant size given at construction. If the writer overtakes the reading of the
    /// characters, the oldest data in the buffer is overwritten, resulting in loss
    /// of data, but preserving being able to read the most recently written characters.
    /// The buffer is designed to be used for example to capture console output
    /// so it can be read by another part of the program or output to a text box.
    /// It locks on an internal object during reading and writing so that it can
    /// be used when the reading and writing processes are running different threads.
    /// </summary>
    public class WritableRingBuffer : TextWriter
    {

        private const int defaultBufferSize = 16384;
        private const int minimumBufferSize = 128;

        private readonly object lockObject = new object();

        public int bufferSize { get; private set; }

        private char[] buffer;
        /// <summary>
        /// The 0-based index in the buffer of the next character to read.
        /// </summary>
        private int nextReadIx = 0;

        /// <summary>
        /// The 0-based index in the buffer of the position where
        /// the next character should be written.
        /// If the value is the same as
        /// <see cref="nextReadIx"/> and the <see cref="atLimit"/> flag is not set, 
        /// then there are no characters available
        /// to read and the buffer is empty.
        /// If the value is numerically less than <see cref="nextReadIx"/>, then
        /// the filled part of the buffer ranges from this index to the end of
        /// the buffer, and again from the start of the buffer up to the index one
        /// less than <see cref="nextReadIx"/>.<br/>
        /// Example:<br/>
        /// Buffer size = 10 (not normally allowed but makes the example easier).<br/>
        /// After construction (. = free, X = data written):<br/>
        /// nextWriteIx == 0   nextReadIx = 0 (..........)<br/>
        /// 5 characters are written.<br/>
        /// nextWriteIx == 5   nextReadIx = 0 (XXXXX.....)<br/>
        /// 4 characters are read<br/>
        /// nextWriteIx == 5   nextReadIx = 4 (....X.....)<br/>
        /// 7 characters are written. The buffer is filled to the end and starts again at the beginning
        /// There are two free spots at index 2 and 3. <br/>
        /// nextWriteIx == 2   nextReadIx = 4 (XX..XXXXXX)<br/>
        /// Another 7 characters are written. The buffer overflows and overwrites 5 characters.
        /// The oldest character is at index 9, but to distinguish this case from the empty buffer the
        /// <see cref="atLimit"/> flag is set.<br/>
        /// nextWriteIx == 9   nextReadIx = 9 (XXXXXXXXXX)<br/>
        /// 10 characters are read. The character at index 9 is returned first, then at indices 0 to 8. 
        /// The <see cref="atLimit"/> flag is reset.<br/>
        /// nextWriteIx == 9   nextReadIx = 9 (..........)<br/>
        /// </summary>
        private int nextWriteIx = 0;

        public bool atLimit { get; private set; } = false;

        public int numberOfCharactersLost { get; private set; } = 0;

        public WritableRingBuffer()
        {
            init(defaultBufferSize);
        }

        /// <summary>
        /// Creates the writable ring buffer with a particular buffer size
        /// </summary>
        /// <param name="bufferSize_">Size of the internal buffer.</param>
        public WritableRingBuffer(int bufferSize_)
        {
            init(bufferSize_);
        }

        public WritableRingBuffer(IFormatProvider formatProvider) : base(formatProvider) 
        {
            init(defaultBufferSize);
        }

        private void init(int bufferSize_)
        {
            if (bufferSize_ < minimumBufferSize)
            {
                bufferSize = minimumBufferSize;
            }
            else
            {
                bufferSize = bufferSize_;
            }
            buffer = new char[bufferSize];
        }

        public override Encoding Encoding 
        {
            get
            {
                return Encoding.UTF8;
            }
        }

        public bool hasCharactersToRead
        {
            get
            {
                lock (lockObject)
                {
                    return ((nextWriteIx != nextReadIx) || atLimit);
                }
            }
        }

        public int numberAvailableToRead
        {
            get
            {
                lock (lockObject)
                {
                    if (nextWriteIx == nextReadIx)
                    {
                        if (atLimit)
                        {
                            return bufferSize;
                        }
                        return 0;
                    }
                    return (nextWriteIx + bufferSize - nextReadIx) % bufferSize; 
                }
            }
        }

        public int numberAvailableToWrite
        {
            get
            {
                lock (lockObject)
                {
                    return bufferSize - numberAvailableToRead; 
                }
            }
        }

        protected override void Dispose(bool disposing)
        {
            buffer = null;
            base.Dispose(disposing);
        }

        public override void Flush()
        {
            lock (lockObject)
            {
                base.Flush(); 
            }
        }

        public override void Write(char value)
        {
            lock (lockObject)
            {
                buffer[nextWriteIx++] = value;
                nextWriteIx = nextWriteIx % bufferSize;
                // Was the buffer overflowed before the write? Then also increment read index
                if (atLimit)
                {
                    nextReadIx = nextWriteIx;
                    numberOfCharactersLost++;
                }
                else
                {
                    // Has the buffer now become overflowed?
                    if (nextWriteIx == nextReadIx)
                    {
                        atLimit = true;
                    }
                } 
            }
        }

        public override void Write(char[] buffer, int index, int count)
        {
            if (buffer == null) throw new ArgumentNullException(nameof(buffer));
            if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
            if (index < 0) throw new ArgumentOutOfRangeException(nameof(index));
            if ((buffer.Length - index) < count) throw new ArgumentException(string.Format("The {3} length == {0} minus {4} == {1} is less than {5} == {2}",
                  buffer.Length, index, count, nameof(buffer), nameof(index), nameof(count)));

            lock (lockObject)
            {
                // Does the request itself already exceed the buffer size?
                if (count > bufferSize)
                {
                    int newStart = index + count - bufferSize;
                    Write(buffer, newStart, bufferSize);
                    atLimit = true;
                    nextReadIx = nextWriteIx;
                    numberOfCharactersLost += (count - bufferSize);
                    return;
                }

                // Space available before starting to write.
                int freeSpacesBeforeStarting = numberAvailableToWrite;

                // First chunk: starting at the write index possibly up to the end of the buffer
                int chunk1Count = Math.Min(count, bufferSize - nextWriteIx);
                Array.Copy(buffer, index, this.buffer, nextWriteIx, chunk1Count);
                int remaining = count - chunk1Count;

                // Anything left to copy?
                if (remaining > 0)
                {
                    int newStart = index + chunk1Count;
                    Array.Copy(buffer, newStart, this.buffer, 0, remaining);
                }

                // New write index
                nextWriteIx = (nextWriteIx + count) % bufferSize;

                // Did the buffer hit its limit?
                if (freeSpacesBeforeStarting <= count)
                {
                    atLimit = true;
                    nextReadIx = nextWriteIx;
                    numberOfCharactersLost += (count - freeSpacesBeforeStarting);
                } 
            }
        }

        public override void Write(string value)
        {
            if (string.IsNullOrEmpty(value)) return;

            int count = value.Length;

            lock (lockObject)
            {
                // Does the request itself already exceed the buffer size?
                if (count > bufferSize)
                {
                    int newStart = count - bufferSize;
                    Write(value.Substring(newStart));
                    atLimit = true;
                    nextReadIx = nextWriteIx;
                    numberOfCharactersLost += (count - bufferSize);
                    return;
                }

                // Space available before starting to write.
                int freeSpacesBeforeStarting = numberAvailableToWrite;

                // First chunk: as many characters as possible up to the end of the buffer
                int chunk1Count = Math.Min(count, bufferSize - nextWriteIx);
                value.CopyTo(0, this.buffer, nextWriteIx, chunk1Count);
                int remaining = count - chunk1Count;

                // Anything left to copy?
                if (remaining > 0)
                {
                    value.CopyTo(chunk1Count, this.buffer, 0, remaining);
                }

                // New write index
                nextWriteIx = (nextWriteIx + count) % bufferSize;

                // Did the buffer hit its limit?
                if (freeSpacesBeforeStarting <= count)
                {
                    atLimit = true;
                    nextReadIx = nextWriteIx;
                    numberOfCharactersLost += (count - freeSpacesBeforeStarting);
                } 
            }
        }

        public string getAllAvailableCharacters()
        {
            lock (lockObject)
            {
                if (!hasCharactersToRead)
                {
                    return string.Empty;
                }

                // Nonfragmented case
                if (nextReadIx < nextWriteIx)
                {
                    string result = new string(buffer, nextReadIx, numberAvailableToRead);
                    atLimit = false;
                    nextReadIx = nextWriteIx;
                    return result;
                }

                // Fragmented case
                StringBuilder sb = new StringBuilder(numberAvailableToRead);
                int chunk1Count = (bufferSize - nextReadIx);
                sb.Append(buffer, nextReadIx, chunk1Count);
                int remaining = numberAvailableToRead - chunk1Count;
                sb.Append(buffer, 0, remaining);
                atLimit = false;
                nextReadIx = nextWriteIx;
                return sb.ToString(); 
            }
        }

    }
}