4
\$\begingroup\$

I created this wrapper to use together with HttpClient streams and ZipArchive. ZipArchive reads .zip index once from the end of the archive, so this wrapper caches last 4MiB of the stream. Also the wrapper avoids pointless seeks until the first read.

I am interested if there any issues with this approach, and whether this can be improved.

namespace Playground
{
    using System;
    using System.IO;
    using System.Net;
    using System.Net.Http;
    using System.Net.Http.Headers;
    using System.Threading;
    using System.Threading.Tasks;

    public class SeekableHttpStream : Stream
    {
        private long _position;
        private long _underlyingStreamOffset;
        private Stream _underlyingStream;
        private bool _forceRequest;

        internal SeekableHttpStream(
            HttpClient client,
            HttpResponseMessage response,
            HttpRequestMessage request)
        {
            Client = client;
            Response = response;
            Request = request;

            var headers = response.Headers;
            var acceptRanges = headers?.AcceptRanges;
            if (acceptRanges == null || !acceptRanges.Contains("bytes"))
            {
                throw new ArgumentException("server does not support HTTP range requests", nameof(request));
            }
            var contentHeaders = response.Content?.Headers;
            if (contentHeaders.ContentLength != null)
            {
                Length = contentHeaders.ContentLength.Value;
            }
            else if (contentHeaders.ContentRange != null)
            {
                if (contentHeaders.ContentRange.Length == null)
                {
                    throw new ArgumentException("missing Content-Range length", nameof(request));
                }

                Length = contentHeaders.ContentRange.Length.Value;
            }
            else
            {
                throw new ArgumentException("failed to determine stream length", nameof(request));
            }
        }

        public HttpClient Client { get; }

        public HttpResponseMessage Response { get; }

        public HttpRequestMessage Request { get; }

        public override bool CanRead => _position < Length;

        public override bool CanSeek => true;

        public override bool CanWrite => false;

        public override long Length { get; }

        public override long Position
        {
            get => _position;
            set => Seek(value, SeekOrigin.Begin);
        }

        public override void Flush()
        {
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            EnsureStreamOpen().GetAwaiter().GetResult();
            int read = _underlyingStream.Read(buffer, offset, count);
            _position += read;
            return read;
        }

        public override int Read(Span<byte> buffer)
        {
            EnsureStreamOpen().GetAwaiter().GetResult();
            int read = _underlyingStream.Read(buffer);
            _position += read;
            return read;
        }

        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            await EnsureStreamOpen(cancellationToken).ConfigureAwait(false);
            int read = await _underlyingStream.ReadAsync(buffer, offset, count, cancellationToken)
                .ConfigureAwait(false);
            _position += read;
            return read;
        }

        public override int ReadByte()
        {
            EnsureStreamOpen().GetAwaiter().GetResult();

            var value = _underlyingStream.ReadByte();
            ++_position;

            return value;
        }

        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
        {
            await EnsureStreamOpen(cancellationToken).ConfigureAwait(false);

            int read = await _underlyingStream.ReadAsync(buffer, cancellationToken)
                 .ConfigureAwait(false);
            _position += read;
            return read;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            return SeekAsync(offset, origin).GetAwaiter().GetResult();
        }

        private ValueTask EnsureStreamOpen(CancellationToken cancellationToken = default)
        {
            if (_underlyingStream == null)
            {
                _forceRequest = true;
                return new ValueTask(SeekAsync(0, SeekOrigin.Current, cancellationToken));
            }

            return default;
        }

        public async Task<long> SeekAsync(long offset, SeekOrigin origin, CancellationToken cancellationToken = default)
        {
            const long SeekThreshold = 1024 * 1024;

            long newPosition = origin switch
            {
                SeekOrigin.Begin => offset,
                SeekOrigin.Current => _position + offset,
                SeekOrigin.End => Length + offset,
                _ => throw new ArgumentOutOfRangeException(nameof(origin)),
            };
            if (newPosition < 0)
            {
                throw new ArgumentOutOfRangeException(nameof(offset));
            }
            if (newPosition > Length)
            {
                throw new NotSupportedException("seeking beyond the length of the stream is not supported");
            }

            long delta = newPosition - _position;
            if (_underlyingStream == null)
            {
                if (_forceRequest)
                {
                    await OpenUnderlyingStream(newPosition, cancellationToken)
                        .ConfigureAwait(false);
                }
                _position = newPosition;
            }
            else if (_underlyingStream.CanSeek && newPosition >= _underlyingStreamOffset && newPosition <= _underlyingStreamOffset + _underlyingStream.Length)
            {
                _underlyingStream.Position = newPosition - _underlyingStreamOffset;
                _position = newPosition;
            }
            else if (delta < 0 || delta > SeekThreshold)
            {
                await OpenUnderlyingStream(newPosition, cancellationToken)
                    .ConfigureAwait(false);
            }
            else if (delta > 0)
            {
                var buffer = new byte[delta];
                await ReadAsync(buffer, 0, (int)delta, cancellationToken);
            }

            return _position;
        }

        private async Task<HttpRequestMessage> CopyHttpRequest()
        {
            var clone = new HttpRequestMessage(Request.Method, Request.RequestUri);

            if (Request.Content != null)
            {
                var bytes = await Request.Content.ReadAsByteArrayAsync()
                    .ConfigureAwait(false);
                clone.Content = new ByteArrayContent(bytes);

                if (Request.Content.Headers != null)
                    foreach (var h in Request.Content.Headers)
                        clone.Content.Headers.Add(h.Key, h.Value);
            }

            clone.Version = Request.Version;

            foreach (var prop in Request.Properties)
            {
                clone.Properties.Add(prop);
            }

            foreach (var header in Request.Headers)
            {
                clone.Headers.TryAddWithoutValidation(header.Key, header.Value);
            }

            return clone;
        }

        private async Task OpenUnderlyingStream(long position, CancellationToken cancellationToken = default)
        {
            const long UseBufferedStreamThreshold = 4 * 1024 * 1024;

            if (position < 0)
            {
                throw new ArgumentOutOfRangeException(nameof(position));
            }

            using var newRequest = await CopyHttpRequest()
                .ConfigureAwait(false);
            if (position > 0)
            {
                var responseHeaders = Response.Headers;
                var contentHeaders = Response.Content.Headers;

                if (responseHeaders.ETag != null)
                {
                    newRequest.Headers.IfRange = new RangeConditionHeaderValue(responseHeaders.ETag);
                }
                else if (contentHeaders.LastModified != null)
                {
                    newRequest.Headers.IfRange = new RangeConditionHeaderValue(contentHeaders.LastModified.Value);
                }

                newRequest.Headers.Range = new RangeHeaderValue(position, null);
            }

            long remainingLength = Length - position;

            var response = await Client.SendAsync(
                newRequest,
                remainingLength <= UseBufferedStreamThreshold ? HttpCompletionOption.ResponseContentRead : HttpCompletionOption.ResponseHeadersRead,
                cancellationToken
            ).ConfigureAwait(false);

            if (!response.IsSuccessStatusCode)
            {
                response.EnsureSuccessStatusCode();
            }
            else if (position > 0 && response.StatusCode != HttpStatusCode.PartialContent)
            {
                response.Dispose();
                throw new InvalidOperationException("range request not supported or content has changed since last request");
            }
            else
            {
                try
                {
                    var stream = await response.Content.ReadAsStreamAsync()
                        .ConfigureAwait(false);
                    if (_underlyingStream != null)
                    {
                        await _underlyingStream.DisposeAsync()
                            .ConfigureAwait(false);
                    }

                    _underlyingStream = stream;
                    _underlyingStreamOffset = position;
                    _forceRequest = false;
                    _position = position;
                }
                catch
                {
                    response.Dispose();
                    throw;
                }
            }
        }

        protected override void Dispose(bool disposing)
        {
            _underlyingStream?.Dispose();
            Response.Dispose();
            Request.Dispose();
            base.Dispose(disposing);
        }

        #region Unsupported write methods

        public override void SetLength(long value)
        {
            throw new NotSupportedException();
        }

        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
        {
            throw new NotSupportedException();
        }

        public override void EndWrite(IAsyncResult asyncResult) => throw new NotSupportedException();

        public override void Write(ReadOnlySpan<byte> buffer) => throw new NotSupportedException();

        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            throw new NotSupportedException();
        }

        public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
        {
            throw new NotSupportedException();
        }

        public override void WriteByte(byte value)
        {
            throw new NotSupportedException();
        }

        public override int WriteTimeout
        {
            get => throw new NotSupportedException();
            set => throw new NotSupportedException();
        }

        #endregion Unsupported write methods
    }

    public static class HttpClientExtensions
    {
        public static async Task<SeekableHttpStream> GetSeekableStreamAsync(this HttpClient client, string requestUri)
        {
            using var request = new HttpRequestMessage(HttpMethod.Get, requestUri);

            return await SendSeekableStreamAsync(client, request)
                .ConfigureAwait(false);
        }

        public static async Task<SeekableHttpStream> GetSeekableStreamAsync(this HttpClient client, Uri requestUri)
        {
            using var request = new HttpRequestMessage(HttpMethod.Get, requestUri);

            return await SendSeekableStreamAsync(client, request)
                .ConfigureAwait(false);
        }

        public static async Task<SeekableHttpStream> SendSeekableStreamAsync(this HttpClient client, HttpRequestMessage request)
        {
            HttpMethod method = request.Method;
            try
            {
                request.Method = HttpMethod.Head;
                var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead)
                    .ConfigureAwait(false);
                response.EnsureSuccessStatusCode();

                try
                {
                    return new SeekableHttpStream(client, response, request);
                }
                catch
                {
                    response.Dispose();
                    throw;
                }
            }
            finally
            {
                request.Method = method;
            }
        }
    }
}
\$\endgroup\$

2 Answers 2

2
\$\begingroup\$
else if (delta > 0)
{
    var buffer = new byte[delta];
    await ReadAsync(buffer, 0, (int)delta, cancellationToken);
}

This code has a fairly nasty bug: ReadAsync is not required to read as much data as you're requesting, it can read less. You either need to use the .Net 7 method ReadExactlyAsync, or you need to read in a loop, until the right number of bytes has been read.

\$\endgroup\$
1
\$\begingroup\$

Update 2023-06-21 So sorry, @svick. When I came across the original post, I happened to be working on a .NET Blazor Web Assembly project, so that is what I wrote the code in. I retested the code in Blazor and it works, but then, like you said, tested in others and it does not work in any .NET Framework or .NET Core versions.

So, if you or anyone else may still need a solution, I think the best that can be done is to wrap the stream in a custom seekable stream. (Note, the .NET BufferedStream will not work for this purpose.) I found this solution in another StackOverlow answer.

I updated my code below to show using that class (but I'm not going to include/duplicate the entire class here).


I didn't try out your code to completely analyze it. I see you have code in there doing a check on the AcceptRanges and ContentRange headers, but I don't know how much of a role those play into your buffering. From my experience and research, the [HttpClient]'s(https://learn.microsoft.com/en-us/dotnet/api/system.net.http.httpclient?view=netframework-4.7.2) response is already seekable. It's buffered by default. There is the MaxResponseContentBufferSize property already available that might effectively be doing what you're trying to accomplish. By default, it will buffer up to 2 GB in memory.

So I can already write the following kind of code:

using var http = new HttpClient();

// Buffer only 10,000, but we'll request 100,000.
http.MaxResponseContentBufferSize = 10000;

var req = new HttpRequestMessage
{
    Method = new HttpMethod("GET"),
    RequestUri = new Uri("https://httpbin.org/bytes/100000"),
};

req.Headers.Add("Accept", "application/octet-stream");

using var res = await http.SendAsync(req, HttpCompletionOption.ResponseHeadersRead);
using var sr = new ReadSeekableStream(await res.Content.ReadAsStreamAsync(), 1000);

// Read the first 100 bytes.
var buf1 = new byte[100];
_ = await sr.ReadAsync(buf1, 0, buf1.Length);

// Then, reset the position back to the beginning to read again.
sr.Position = 0;

// Read the first 100 bytes again.
var buf2 = new byte[100];
_ = await sr.ReadAsync(buf2, 0, buf2.Length);

// Prove that both reads contain the same data, which should be a good indication that
// the stream is seekable and doing it correctly.
var eq = buf1.SequenceEqual(buf2);

Console.WriteLine($"Are equal: {eq}");

// Reset again and read the entire response.
sr.Position = 0;

// However you want to go about reading all of the data and what to do with it
// is up to you.
var totalLength = 0;
var buf = new byte[8192];
var count = 0;
while ((count = await sr.ReadAsync(buf, 0, buf.Length)) > 0)
{
    totalLength += count;
}

// We should now get all 100,000 bytes.
Console.WriteLine($"All length: {totalLength}");

Is that what you're trying to accomplish with your code?

\$\endgroup\$
4
  • \$\begingroup\$ HttpClient won't do any buffering because you specified HttpCompletionOption.ResponseHeadersRead. \$\endgroup\$
    – Nazar554
    Nov 12, 2022 at 21:27
  • \$\begingroup\$ It's not that it won't do any buffering. That flag simply prevents it from trying to download and buffer the entire response before returning from the method call. Did you try my sample code? It runs and outputs true for the comparison, indicating that the stream is seekable and is therefore buffered. Again, there is a difference between the Response.Content being buffered, and being completely downloaded before the call to await http.SendAsync() completes and control is returned to your code. \$\endgroup\$
    – Kevin
    Nov 14, 2022 at 17:41
  • \$\begingroup\$ At least on .Net 7, this code throws an exception, because the stream is not seekable. \$\endgroup\$
    – svick
    Jun 20, 2023 at 13:19
  • \$\begingroup\$ @svick, so sorry. I updated my answer post with a comment for you. \$\endgroup\$
    – Kevin
    Jun 22, 2023 at 1:23

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.