c# - FIFO/Queue buffer specialising in byte streams

Question

Is there any .NET data structure/combination of classes that allows for byte data to be appended to the end of a buffer but all peeks and reads are from the start, shortening the buffer when I read?

The MemoryStream class seems to do part of this, but I need to maintain separate locations for reading and writing, and it doesn't automatically discard the data at the start after it's read.

An answer has been posted in reply to this question which is basically what I'm trying to do but I'd prefer something I can do asynchronous I/O on in different components of the same process, just like a normal pipe or even a network stream (I need to filter/process the data first).

Answer 1

I'll post a stripped out copy of some logic i wrote for a project at work once. The advantage of this version is that it works with a linked list of buffered data and therefore you dont have to cache huge amounts of memory and/or copy memory around when reading. furthermore, its thread safe and behaves like a network stream, that is: When reading when there is no data available: Wait untill there is data available or timeout. Also, when reading x amounts of bytes and there are only y amounts of bytes, return after reading all bytes. I hope this helps!

    public class SlidingStream : Stream
{
    #region Other stream member implementations

    ...

    #endregion Other stream member implementations

    public SlidingStream()
    {
        ReadTimeout = -1;
    }

    private readonly object _writeSyncRoot = new object();
    private readonly object _readSyncRoot = new object();
    private readonly LinkedList<ArraySegment<byte>> _pendingSegments = new LinkedList<ArraySegment<byte>>();
    private readonly ManualResetEventSlim _dataAvailableResetEvent = new ManualResetEventSlim();

    public int ReadTimeout { get; set; }

    public override int Read(byte[] buffer, int offset, int count)
    {
        if (_dataAvailableResetEvent.Wait(ReadTimeout))
            throw new TimeoutException("No data available");

        lock (_readSyncRoot)
        {
            int currentCount = 0;
            int currentOffset = 0;

            while (currentCount != count)
            {
                ArraySegment<byte> segment = _pendingSegments.First.Value;
                _pendingSegments.RemoveFirst();

                int index = segment.Offset;
                for (; index < segment.Count; index++)
                {
                    if (currentOffset < offset)
                    {
                        currentOffset++;
                    }
                    else
                    {
                        buffer[currentCount] = segment.Array[index];
                        currentCount++;
                    }
                }

                if (currentCount == count)
                {
                    if (index < segment.Offset + segment.Count)
                    {
                        _pendingSegments.AddFirst(new ArraySegment<byte>(segment.Array, index, segment.Offset + segment.Count - index));
                    }
                }

                if (_pendingSegments.Count == 0)
                {
                    _dataAvailableResetEvent.Reset();

                    return currentCount;
                }
            }

            return currentCount;
        }
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        lock (_writeSyncRoot)
        {
            byte[] copy = new byte[count];
            Array.Copy(buffer, offset, copy, 0, count);

            _pendingSegments.AddLast(new ArraySegment<byte>(copy));

            _dataAvailableResetEvent.Set();
        }   
    }
}