3
\$\begingroup\$

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;
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ The following fails for me with an ArgumentOutOfRangeException on req.AddRange: using (var stream = new PartialHTTPStream("http://google.com")) { using (var reader = new StreamReader(stream)) { Console.WriteLine(reader.ReadToEnd()); } }. Am I using it wrong? Can you provide sample usage? \$\endgroup\$ – mjolka Nov 24 '14 at 3:27
  • \$\begingroup\$ There was a bug where "cacheRemaining -= (int)delta;" was misplaced. Please try again. \$\endgroup\$ – Reinderien Nov 24 '14 at 3:32
  • \$\begingroup\$ My first thought is that you should split this into two classes, one which does the caching and one which does wrap http. SRP. \$\endgroup\$ – CodesInChaos Nov 27 '14 at 14:40
  • \$\begingroup\$ BufferedStream might already implement the desired caching behaviour. \$\endgroup\$ – CodesInChaos Nov 27 '14 at 14:52
3
\$\begingroup\$

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);
    }
} 
\$\endgroup\$
  • \$\begingroup\$ if (position == value) { return; } - true. Potential "bug" (you should enable spellcheck..) - this is not a bug. If stream is null, that means that cacheRemaining == 0, so that condition will never be hit. Stream.Close() - true. As to the caching - it is, in fact, caching, where the cache is kept within the request itself. This is caching in the sense of not having to re-seek and being able to read from the existing response if it's within the requested range. \$\endgroup\$ – Reinderien Nov 25 '14 at 1:45
  • \$\begingroup\$ Though (some of) these suggestions are useful, in general I'm looking for a suggestion on how to better use built-in features of .NET instead of having to do this all myself; thus will keep this open to other answers. \$\endgroup\$ – Reinderien Nov 25 '14 at 2:08
  • \$\begingroup\$ "The Read() method shows unexpected behaviour" - no, it doesn't. If too many bytes are requested, the method can simply read fewer bytes and return that count. \$\endgroup\$ – Reinderien Nov 27 '14 at 2:33
  • 2
    \$\begingroup\$ abstract Stream class ArgumentException if "The sum of offset and count is larger than the buffer length." FileStream ArgumentException if "offset and count describe an invalid range in array" which comes down to if (array.Length - offset < count) \$\endgroup\$ – Heslacher Nov 27 '14 at 5:55
1
\$\begingroup\$

@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;
        }
    }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Well I don't see the drawbacks you mentioned. Using the overloaded constructor will invalidate the first points. Btw. Your constructor lacks checking cacheLen where one could pass a negative value. \$\endgroup\$ – Heslacher Oct 2 '18 at 16:00
  • \$\begingroup\$ @Heslacher you use fixed start cache position. E.g. try Position = 1020; Read(..., 0, 16). Ctor fixed. \$\endgroup\$ – xmedeko Oct 3 '18 at 7:37
1
\$\begingroup\$

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.
...

\$\endgroup\$
  • \$\begingroup\$ Is this distinct from the concerns regarding implementation conformality presented in Heslacher's answer? \$\endgroup\$ – VisualMelon Oct 3 '18 at 16:52
  • \$\begingroup\$ @VisualMelon, now that you mention it I can see it. I had previously skim-read the answer without seeing it. \$\endgroup\$ – Peter Taylor Oct 3 '18 at 19:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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