BufferedBytesFromStreamProducer.cs

Go to the documentation of this file.

using System;
using System.IO;
 
namespace AnalyzeRe.LargeDataUpload
{
    public class BufferedBytesFromStreamProducer : AdaptiveBufferedBytesProducer
    {
        private const int SourceBufferReadIncrements = 8;
 
        #region Public Properties
        public bool EndOfStreamDetected { get; private set; }
 
        public long? BytesToRead { get; }
        #endregion Public Properties
 
        #region Private Properties
        private readonly Stream _source;
        #endregion Private Properties
 
        #region Constructor
        public BufferedBytesFromStreamProducer(Stream data_source, long? bytes_to_read = null,
            int? min_buffer_length = null, int? max_buffer_length = null)
            : base(min_buffer_length, max_buffer_length)
        {
            _source = data_source;
            BytesToRead = bytes_to_read ?? (_source.CanSeek ? _source.Length - _source.Position : (long?)null);
            // Don't bother allocating buffers larger than the number of bytes to read
            if (BytesToRead.HasValue && (!max_buffer_length.HasValue || BytesToRead < max_buffer_length))
                MaximumBufferLength = (int)BytesToRead.Value;
        }
        #endregion Constructor
 
        #region Private Methods
        protected override bool IsProducerFinished()
        {
            return EndOfStreamDetected || (BytesToRead.HasValue && TotalBytesRead == BytesToRead);
        }
 
        protected override bool TryProduceNext(out BufferedBytes next, int buffer_size)
        {
            next = new BufferedBytes();
            if (EndOfStreamDetected)
                return false;
 
            // If the target buffer length is greater than the remaining number of bytes
            // we've been asked to read from the input stream, just read the remaining bytes.
            if (BytesToRead.HasValue && BytesToRead.Value - TotalBytesRead < buffer_size)
                buffer_size = (int)(BytesToRead.Value - TotalBytesRead);
 
            next.Bytes = new byte[buffer_size];
            next.LengthFilled = 0;
 
            // Read bytes from the source in increments of "MinimumBufferLength" until we reach
            // the allocated buffer length, or the consumer starts waiting on us for more bytes.
            while (!EndOfStreamDetected && next.LengthFilled < buffer_size &&
                (next.LengthFilled < MinimumBufferLength || !IsConsumerWaiting))
            {
                // Request no fewer than MinimumBufferLength, but if the buffer size is greater than
                // the minimum buffer length, fill it in increments to give us a chance to detect
                // if the consumer is ready for more bytes before we've filled the entire buffer.
                int bytes_to_read = Math.Max(MinimumBufferLength, buffer_size / SourceBufferReadIncrements);
                // We cannot request more bytes than there is room left in the destination buffer.
                bytes_to_read = Math.Min(bytes_to_read, buffer_size - next.LengthFilled);
 
                int read = _source.Read(next.Bytes, next.LengthFilled, bytes_to_read);
                next.LengthFilled += read;
                if (read == 0 || !_source.CanRead)
                    EndOfStreamDetected = true;
            }
            if (!_source.CanRead)
                EndOfStreamDetected = true;
 
            next.TotalBytesRead = TotalBytesRead + next.LengthFilled;
            // If we are returning more than 0 bytes, return success
            if (next.LengthFilled > 0)
                return true;
            // If we've reached the end of the stream, but we haven't read as many bytes as we
            // were expecting, raise an error.
            if (EndOfStreamDetected && BytesToRead.HasValue && TotalBytesRead > BytesToRead)
                throw new Exception("The source stream is reporting that it has no more data " +
                    "(CanRead is false), but we have only read " + TotalBytesRead + " bytes and " +
                    "were instructed at construction time to read exactly " + BytesToRead + " bytes.");
            return false;
        }
        #endregion Private Methods
    }
}