This stream class is designed to hold data only until it is read from the stream. As opposed to the MemoryStream which holds data until the stream is closed. It was written some time ago and was based on a sample found online. I can no longer find the source, but I believe I used it as a start and cleaned it up.
I am interested in reviewing the performance both in terms of speed and memory usage (there are a number of spots that create byte arrays and copy data from one array to another).
/// <summary>
/// This stream maintains data only until the data is read, then it is purged from the stream.
/// </summary>
public class MemoryQueueBufferStream : Stream, IDisposable
{
//Maintains the streams data. The Queue object provides an easy and efficient way to add and remove data
//Each item in the queue represents each write to the stream. Every call to write translates to an item in the queue
private Queue<byte[]> lstBuffers_m;
//If the item on the top of the queue is read, but the amount read is less then the size of the data in the queue,
//the remaining data is put in this buffer because the item from the queue is removed. Subsequent read calls will first read
//from this buffer until it is depleted, then the queue will be read again.
private byte[] binLeftOverFromFirstBuffer_m = null;
private bool bIsOpen_m;
public MemoryQueueBufferStream()
{
this.bIsOpen_m = true;
this.lstBuffers_m = new Queue<byte[]>();
}
/// <summary>
/// Reads up to count bytes from the stream, and removes the read data from the stream.
/// </summary>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="count"></param>
/// <returns></returns>
public override int Read(byte[] buffer, int offset, int count)
{
this.ValidateBufferArgs(buffer, offset, count);
if (!this.bIsOpen_m)
{
throw new ObjectDisposedException(null, this.GetType().Name + " is closed");
}
int iRemainingBytesToRead = count;
//If there was data left over return it now before reading from the queue
if (binLeftOverFromFirstBuffer_m != null)
{
int iBytesRead = ReadFromBufferAndTrim(ref binLeftOverFromFirstBuffer_m, buffer, offset, iRemainingBytesToRead);
offset += iBytesRead;
iRemainingBytesToRead -= iBytesRead;
}
while ((iRemainingBytesToRead > 0) && (lstBuffers_m.Count > 0))
{
//Get data from the queue, and remove it from the queue, and put it in a buffer
byte[] binNextBuffer = this.lstBuffers_m.Dequeue();
int bytesRead = ReadFromBufferAndTrim(ref binNextBuffer, buffer, offset, iRemainingBytesToRead);
offset += bytesRead;
iRemainingBytesToRead -= bytesRead;
if (binNextBuffer != null)
{
// If the buffer contains more data than requested, we need to hold it for subsequent reads.
// ReadFromBufferAndTrim has removed the part of the buffer that was already read. The remaining part will
// be read at the next call (and possibly trimmed again)
// before going to the next buffer in the queue.
this.binLeftOverFromFirstBuffer_m = binNextBuffer;
}
}
int totalBytesRead = count - iRemainingBytesToRead;
return totalBytesRead;
}
private static int ReadFromBufferAndTrim(ref byte[] srcBuffer, byte[] destBuffer, int destOffset, int remainingBytesToRead)
{
int bytesToRead = Math.Min(srcBuffer.Length, remainingBytesToRead);
Buffer.BlockCopy(srcBuffer, 0, destBuffer, destOffset, bytesToRead);
if (bytesToRead < srcBuffer.Length)
{
srcBuffer = TrimBufferStart(srcBuffer, bytesToRead);
}
else
{
srcBuffer = null;
}
return bytesToRead;
}
private static byte[] TrimBufferStart(byte[] buffer, int firstByteToKeep)
{
int trimmedLength = buffer.Length - firstByteToKeep;
byte[] trimmedBuffer = new byte[trimmedLength];
Buffer.BlockCopy(buffer, firstByteToKeep, trimmedBuffer, 0, trimmedLength);
return trimmedBuffer;
}
private void ValidateBufferArgs(byte[] buffer, int offset, int count)
{
if (offset < 0)
{
throw new ArgumentOutOfRangeException("offset", "offset must be non-negative");
}
if (count < 0)
{
throw new ArgumentOutOfRangeException("count", "count must be non-negative");
}
if ((buffer.Length - offset) < count)
{
throw new ArgumentException("requested count exceeds available size");
}
}
/// <summary>
/// Writes data to the stream
/// </summary>
/// <param name="buffer">Data to copy into the stream</param>
/// <param name="offset"></param>
/// <param name="count"></param>
public override void Write(byte[] buffer, int offset, int count)
{
this.ValidateBufferArgs(buffer, offset, count);
if (!bIsOpen_m)
{
throw new ObjectDisposedException(null, this.GetType().Name + " is closed");
}
byte[] newBuffer = new byte[count];
Buffer.BlockCopy(buffer, offset, newBuffer, 0, count);
//Add the data to the queue
this.lstBuffers_m.Enqueue(newBuffer);
}
public override bool CanSeek
{
get { return false; }
}
public override long Position
{
get
{
throw new NotSupportedException(this.GetType().Name + " is not seekable");
}
set
{
throw new NotSupportedException(this.GetType().Name + " is not seekable");
}
}
public override bool CanWrite
{
get { return true; }
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException(this.GetType().Name + " is not seekable");
}
public override void SetLength(long value)
{
throw new NotSupportedException(this.GetType().Name + " length can not be changed");
}
public override bool CanRead
{
get { return true; }
}
public override long Length
{
get
{
return this.lstBuffers_m.Sum(b => b.Length) + this.binLeftOverFromFirstBuffer_m.Length;
}
}
public override void Flush()
{
}
#region IDisposable Members
void IDisposable.Dispose()
{
this.lstBuffers_m.Clear();
this.lstBuffers_m = null;
this.binLeftOverFromFirstBuffer_m = null;
this.bIsOpen_m = false;
this.Close();
}
#endregion
}
