BufferedBytesResizer.cs

Go to the documentation of this file.

using System;
 
namespace AnalyzeRe.LargeDataUpload
{
    public class BufferedBytesResizer : AdaptiveBufferedBytesProducer
    {
        #region Public Properties
        public bool EndOfStreamDetected { get; private set; }
        #endregion Public Properties
 
        #region Private Properties
        private readonly IProducerConsumerBuffer<BufferedBytes> _source;
 
        private byte[] _leftOverBytes;
 
        private int _leftOverStart;
 
        private int _leftOverLength;
        #endregion Private Properties
 
        #region Constructor
        public BufferedBytesResizer(IProducerConsumerBuffer<BufferedBytes> data_source,
            int? min_buffer_length = null, int? max_buffer_length = null)
            : base(min_buffer_length, max_buffer_length)
        {
            _source = data_source;
        }
        #endregion Constructor
 
        #region Private Methods
        protected override void OnStart()
        {
            base.OnStart();
            // Start the source producer if it hasn't already started.
            if (!_source.IsRunning)
                _source.Start(Cancellation.Token);
        }
 
        protected override bool IsProducerFinished() =>
           EndOfStreamDetected && _leftOverBytes == null;
 
        protected override bool TryProduceNext(out BufferedBytes next, int buffer_size)
        {
            next = new BufferedBytes { Bytes = new byte[buffer_size] };
            // First consume leftOver bytes, but only as many as can fit in the next buffer.
            int bytes_to_read = Math.Min(buffer_size, _leftOverLength);
            // If there are leftover bytes, consume them first
            if (_leftOverBytes != null)
            {
                Array.Copy(_leftOverBytes, _leftOverStart, next.Bytes, 0, bytes_to_read);
                next.LengthFilled = bytes_to_read;
                // If we've filled the buffer and still have stuff left over,
                // account for the left over amount we've consumed and return.
                if (bytes_to_read < _leftOverLength)
                {
                    _leftOverStart += bytes_to_read;
                    _leftOverLength -= bytes_to_read;
                    return true;
                }
                _leftOverBytes = null;
            }
            // Get the next buffer from the source until the buffer is full or
            // or the source stream has run out of bytes, or until the consumer starts
            // waiting for more bytes (if we've exceeded the minimum bytes)
            while (_source.CanTake && next.LengthFilled < buffer_size &&
                (next.LengthFilled < MinimumBufferLength || !IsConsumerWaiting))
            {
                BufferedBytes source_buffer = _source.TryTake(out bool success, Cancellation.Token);
                if (!success)
                    break;
 
                // We cannot request more bytes than there is room left in the destination buffer
                // We cannot request more bytes than there are in the source buffer
                bytes_to_read = Math.Min(buffer_size - next.LengthFilled, source_buffer.LengthFilled);
 
                Array.Copy(source_buffer.Bytes, 0, next.Bytes, next.LengthFilled, bytes_to_read);
                next.LengthFilled += bytes_to_read;
                next.TotalBytesRead = source_buffer.TotalBytesRead;
 
                // If we didn't use all of the source buffer, save it for later
                if (source_buffer.LengthFilled > bytes_to_read)
                {
                    _leftOverBytes = source_buffer.Bytes;
                    _leftOverStart = bytes_to_read;
                    _leftOverLength = source_buffer.LengthFilled - bytes_to_read;
                    next.TotalBytesRead -= _leftOverLength;
                    break;
                }
            }
            if (next.LengthFilled == 0 || !_source.CanTake)
                EndOfStreamDetected = true;
            return next.LengthFilled != 0;
        }
        #endregion Private Methods
    }
}