Seekable HTTP Range Stream

Question

I read around for a while and surprisingly didn't find a fitting solution. I need a .NET Stream that internally issues HTTP Range requests when it seeks. There are huge files server-side for which I only need small amounts of data in a very random-access way. The consumer of the custom stream (beyond my control) issues lots of tiny, sequential reads, so the stream has to do caching. I have the following, and it seems to work quite well, but it's long and it's irritating my sense of "this should be easier in .NET".

Are there any suggestions to improve or shrink this?

class PartialHTTPStream : Stream, IDisposable
{
    Stream stream;
    WebResponse resp;
    int cacheRemaining = 0;
    const int cachelen = 1024;

    public string Url { get; private set; }
    public override bool CanRead { get { return true; } }
    public override bool CanWrite { get { return false; } }
    public override bool CanSeek { get { return true; } }

    long position = 0;
    public override long Position
    {
        get { return position; }
        set
        {
            long delta = value - position;
            if (delta == 0)
                return;
            if (delta > 0 && delta < cacheRemaining)
            {
                Console.WriteLine("Seeking in cache");
                byte[] dummy = new byte[delta];
                cacheRemaining -= (int)delta;
                while (delta > 0)
                {
                    int nread = stream.Read(dummy, 0, (int)delta);
                    if (nread == 0) throw new IOException();
                    delta -= nread;
                }
            }
            else cacheRemaining = 0;
            position = value;
            Console.WriteLine("Seek {0}", value);
        }
    }

    long? length;
    public override long Length
    {
        get
        {
            if (length == null)
            {
                HttpWebRequest request = HttpWebRequest.CreateHttp(Url);
                request.Method = "HEAD";
                length = request.GetResponse().ContentLength;
            }
            return length.Value;
        }
    }

    public PartialHTTPStream(string Url) { this.Url = Url; }

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

    public override int Read(byte[] buffer, int offset, int count)
    {
        if (cacheRemaining == 0)
        {
            Console.WriteLine("Cache miss");
            if (stream != null)
            {
                stream.Close();
                resp.Close();
            }
            HttpWebRequest req = HttpWebRequest.CreateHttp(Url);
            cacheRemaining = (int)Math.Min(Length - Position, Math.Max(count, cachelen));
            req.AddRange(Position, Position + cacheRemaining - 1);
            resp = req.GetResponse();
            stream = resp.GetResponseStream();
        }

        count = Math.Min(buffer.Length - offset, Math.Min(cacheRemaining, count));
        Console.WriteLine("Read {0} @ {1}", count, Position);

        int nread = stream.Read(buffer, offset, count);
        position += nread;
        cacheRemaining -= nread;
        return nread;
    }

    public override void Write(byte[] buffer, int offset, int count)
    { throw new NotImplementedException(); }

    public override long Seek(long pos, SeekOrigin origin)
    {
        switch (origin)
        {
            case SeekOrigin.End:
                Position = Length + pos;
                break;
            case SeekOrigin.Begin:
                Position = pos;
                break;
            case SeekOrigin.Current:
                Position += pos;
                break;
        }
        return Position;
    }

    public override void Flush() { }

    new void Dispose()
    {
        base.Dispose();
        if (stream != null)
        {
            stream.Dispose();
            stream = null;
        }
        if (resp != null)
        {
            resp.Dispose();
            resp = null;
        }
    }
}

Answer 1

Position.Set

This

long delta = value - position;
if (delta == 0)
    return;

isn't understandable at first glance. You should switch it to

if (position == value) { return; }
long delta = value - position;

You declare long delta for the above substraction but later on you will at least 2 times cast it to int. It would be better to do it the other way.

if (position == value) { return;}
int delta = (int)(value - position);  

Potential "bug"

Assume the user of this class will call either Seek() or Position.Set() your code will throw an exception because stream == null.

You should extract the initializing/opening of the stream to a separate method which you call from the constructor, and if needed (cacheRemaining == 0) from the Read() method.

private void OpenStream(int count = 0)
{
    if (stream != null) { resp.Close(); }

    HttpWebRequest req = HttpWebRequest.CreateHttp(Url);

    cacheRemaining = (int)Math.Min(Length - Position, Math.Max(count, cachelen));
    req.AddRange(Position, Position + cacheRemaining - 1);

    resp = req.GetResponse();
    stream = resp.GetResponseStream();

}

Stream.Close()

The call to Stream.Close is not neccessary if you are closing the WebResponse as this is automatically closing the underlaying stream.

Naming

Shortening variable names to resp does not help for readability. Simply name it response.

Variables should be named using camelCase casing which you do except for cachelen which should be cacheLen or cacheLength.

General

What I don't see in your code is the caching. You are using cacheRemaining and cachelen but you don't cache anything.

The Read() method shows unexpected behaviour. Passing in an array with Length == 10 and an offset == 5 then with count == 6 this should throw an exception.

Using more of the built-in features of .NET

For sure that can be done, but as there is no seekable Network or HTTP stream you need to write some amount of code, but I don't see where this should matter, as you only do this once.

What would I do different to your implementation ?

• copy the responsestream to a MemoryStream to take advantage of using the Seek() method.
• overwrite the Close() method instead of implementing IDisposable as the Stream's Dispose() method only calls the Close() method.
• make the cacheLength adjustable, at least by using a constructor parameter.
• Ensure that the behaviour of my implementation is the same as of the abstract Stream class
  
  • Each method/property should throw an ObjectDisposedException if the stream is disposed
  • Read() method
    1. If buffer is null throw an ArgumentNullException
    2. If offset or count is negativ throw an ArgumentOutOfRangeException
    3. If buffer.Length - offset < count throw an ArgumentException

But as always code does tell more than just words

class PartialHTTPStream : Stream
{
    private readonly Int64 cacheLength = 1024;
    private const Int32 noDataAvaiable = 0;
    private MemoryStream stream = null;
    private Int64 currentChunkNumber = -1;
    private Int64? length;
    private Boolean isDisposed = false;

    public PartialHTTPStream(String url)
        : this(url, 1024) { }

    public PartialHTTPStream(String url, Int64 cacheLength)
    {
        if (cacheLength > 0) { this.cacheLength = cacheLength; }
        Url = url;
    }

    public String Url { get; private set; }

    public override Boolean CanRead
    {
        get
        {
            EnsureNotDisposed();
            return true;
        }
    }

    public override Boolean CanWrite
    {
        get
        {
            EnsureNotDisposed();
            return false;
        }
    }

    public override Boolean CanSeek
    {
        get
        {
            EnsureNotDisposed();
            return true;
        }
    }

    public override Int64 Length
    {
        get
        {
            EnsureNotDisposed();
            if (length == null)
            {
                HttpWebRequest request = HttpWebRequest.CreateHttp(Url);
                request.Method = "HEAD";
                length = request.GetResponse().ContentLength;
            }
            return length.Value;
        }
    }

    public override Int64 Position
    {
        get
        {
            EnsureNotDisposed();
            Int64 streamPosition = (stream != null) ? stream.Position : 0;
            Int64 position = (currentChunkNumber != -1) ? currentChunkNumber * cacheLength : 0;

            return position + streamPosition;
        }
        set
        {
            EnsureNotDisposed();
            EnsurePositiv(value, "Position");
            Seek(value);
        }
    }

    public override Int64 Seek(Int64 offset, SeekOrigin origin)
    {
        EnsureNotDisposed();
        switch (origin)
        {
            case SeekOrigin.Begin:
                break;
            case SeekOrigin.Current:
                offset = Position + offset;
                break;
            default:
                offset = Length + offset;
                break;
        }

        return Seek(offset);
    }

    private Int64 Seek(Int64 offset)
    {
        Int64 chunkNumber = offset / cacheLength;

        if (currentChunkNumber != chunkNumber)
        {
            ReadChunk(chunkNumber);
            currentChunkNumber = chunkNumber;
        }

        offset = offset - currentChunkNumber * cacheLength;

        stream.Seek(offset, SeekOrigin.Begin);

        return Position;
    }

    private void ReadNextChunk()
    {
        currentChunkNumber += 1;
        ReadChunk(currentChunkNumber);
    }

    private void ReadChunk(Int64 chunkNumberToRead)
    {
        Int64 rangeStart = chunkNumberToRead * cacheLength;

        if (rangeStart > Length) { return; }

        Int64 rangeEnd = rangeStart + cacheLength - 1;
        if (rangeStart + cacheLength > Length)
        {
            rangeEnd = Length - 1;
        }

        if (stream != null) { stream.Close(); }
        stream = new MemoryStream((int)cacheLength);

        HttpWebRequest request = HttpWebRequest.CreateHttp(Url);
        request.AddRange(rangeStart, rangeEnd);

        using (WebResponse response = request.GetResponse())
        {
            response.GetResponseStream().CopyTo(stream);
        }

        stream.Position = 0;
    }

    public override void Close()
    {
        EnsureNotDisposed();

        base.Close();
        if (stream != null) { stream.Close(); }
        isDisposed = true;
    }

    public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count)
    {
        EnsureNotDisposed();

        EnsureNotNull(buffer, "buffer");
        EnsurePositiv(offset, "offset");
        EnsurePositiv(count, "count");

        if (buffer.Length - offset < count) { throw new ArgumentException("count"); }

        if (stream == null) { ReadNextChunk(); }

        if (Position >= Length) { return noDataAvaiable; }

        if (Position + count > Length)
        {
            count = (Int32)(Length - Position);
        }

        Int32 bytesRead = stream.Read(buffer, offset, count);
        Int32 totalBytesRead = bytesRead;
        count -= bytesRead;

        while (count > noDataAvaiable)
        {
            ReadNextChunk();
            offset = offset + bytesRead;
            bytesRead = stream.Read(buffer, offset, count);
            count -= bytesRead;
            totalBytesRead = totalBytesRead + bytesRead;
        }

        return totalBytesRead;

    }

    public override void SetLength(Int64 value)
    {
        EnsureNotDisposed();
        throw new NotImplementedException();
    }

    public override void Write(Byte[] buffer, Int32 offset, Int32 count)
    {
        EnsureNotDisposed();
        throw new NotImplementedException();
    }

    public override void Flush()
    {
        EnsureNotDisposed();
    }

    private void EnsureNotNull(Object obj, String name)
    {
        if (obj != null) { return; }
        throw new ArgumentNullException(name);
    }
    private void EnsureNotDisposed()
    {
        if (!isDisposed) { return; }
        throw new ObjectDisposedException("PartialHTTPStream");
    }
    private void EnsurePositiv(Int32 value, String name)
    {
        if (value > -1) { return; }
        throw new ArgumentOutOfRangeException(name);
    }
    private void EnsurePositiv(Int64 value, String name)
    {
        if (value > -1) { return; }
        throw new ArgumentOutOfRangeException(name);
    }
    private void EnsureNegativ(Int64 value, String name)
    {
        if (value < 0) { return; }
        throw new ArgumentOutOfRangeException(name);
    }
} 

Answer 2

@Heslacher's solution has drawbacks:

• Large data are read by chunks.
• When reading small data on a chunk boundary, two HTTP requests are issued.
• Length is not necessary when you do not need it. E.g. end of resource may be detected by Read returning less then requested bytes.

Optimized solution:

public class PartialHttpStream : Stream, IDisposable
{
    private const int CacheLen = 8192;

    // Cache for short requests.
    private readonly byte[] cache;
    private readonly int cacheLen;
    private Stream stream;
    private WebResponse response;
    private long position = 0;
    private long? length;
    private long cachePosition;
    private int cacheCount;

    public PartialHttpStream(string url, int cacheLen = CacheLen)
    {
        if (string.IsNullOrEmpty(url))
            throw new ArgumentException("url empty");
        if (cacheLen <= 0)
            throw new ArgumentException("cacheLen must be greater than 0");

        Url = url;
        this.cacheLen = cacheLen;
        cache = new byte[cacheLen];
    }

    public string Url { get; private set; }

    public override bool CanRead { get { return true; } }
    public override bool CanWrite { get { return false; } }
    public override bool CanSeek { get { return true; } }

    public override long Position {get; set;}

    /// <summary>
    /// Lazy initialized length of the resource.
    /// </summary>
    public override long Length
    {
        get
        {
            if (length == null)
                length = HttpGetLength();
            return length.Value;
        }
    }

    /// <summary>
    /// Count of HTTP requests. Just for statistics reasons.
    /// </summary>
    public int HttpRequestsCount { get; private set; }

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

    public override int Read(byte[] buffer, int offset, int count)
    {
        if (buffer == null)
            throw new ArgumentNullException(nameof(buffer));
        if (offset < 0 || offset >= buffer.Length)
            throw new ArgumentException(nameof(offset));
        if (count < 0 || offset + count > buffer.Length)
            throw new ArgumentException(nameof(count));

        long curPosition = Position;
        Position += ReadFromCache(buffer, ref offset, ref count);
        //if (Position > curPosition)
        //    Console.WriteLine($"Cache hit {Position - curPosition}");
        if (count > cacheLen)
        {
            // large request, do not cache
            Position += HttpRead(buffer, offset, count);
        }
        else if (count > 0)
        {
            // read to cache
            cachePosition = Position;
            cacheCount = HttpRead(cache, 0, cacheLen);
            Position += ReadFromCache(buffer, ref offset, ref count);
        }

        return (int)(Position - curPosition);
    }

    public override void Write(byte[] buffer, int offset, int count)
    { throw new NotImplementedException(); }

    public override long Seek(long pos, SeekOrigin origin)
    {
        switch (origin)
        {
            case SeekOrigin.End:
                Position = Length + pos;
                break;

            case SeekOrigin.Begin:
                Position = pos;
                break;

            case SeekOrigin.Current:
                Position += pos;
                break;
        }
        return Position;
    }

    public override void Flush()
    {
    }

    private int ReadFromCache(byte[] buffer, ref int offset, ref int count)
    {
        if (cachePosition > Position || (cachePosition + cacheCount) <= Position)
            return 0; // cache miss
        int ccOffset = (int)(Position - cachePosition);
        int ccCount = Math.Min(cacheCount - ccOffset, count);
        Array.Copy(cache, ccOffset, buffer, offset, ccCount);
        offset += ccCount;
        count -= ccCount;
        return ccCount;
    }

    private int HttpRead(byte[] buffer, int offset, int count)
    {
        HttpRequestsCount++;
        HttpWebRequest req = HttpWebRequest.CreateHttp(Url);
        req.AddRange(Position, Position + count - 1);
        try
        {
            response = req.GetResponse();
        }
        catch (WebException ex)
        {
            var response = (HttpWebResponse)ex.Response;
            if (response.StatusCode == HttpStatusCode.RequestedRangeNotSatisfiable)
                return 0;
            throw ex;
        }
        int nread;
        using (Stream stream = response.GetResponseStream())
            nread = stream.Read(buffer, offset, count);
        return nread;
    }

    private long HttpGetLength()
    {
        HttpRequestsCount++;
        HttpWebRequest request = HttpWebRequest.CreateHttp(Url);
        request.Method = "HEAD";
        return request.GetResponse().ContentLength;
    }

    private new void Dispose()
    {
        base.Dispose();
        if (stream != null)
        {
            stream.Dispose();
            stream = null;
        }
        if (response != null)
        {
            response.Dispose();
            response = null;
        }
    }
}

Answer 3

This is a minor point, but

    public override int Read(byte[] buffer, int offset, int count)
    {
        ...
        count = Math.Min(buffer.Length - offset, Math.Min(cacheRemaining, count));
        ...
    }

violates the contract of Stream.Read:

Exceptions

ArgumentException
The sum of offset and count is larger than the buffer length.
...