MeteorLight/http.go

package main

import (
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"runtime"
	"strconv"
)

const rangeReaderBufferSize = 4096

type RangeReadSeekCloser struct {
	uri    *url.URL
	size   int64
	i      int64
	body   io.ReadCloser
	buf    []byte
	ib     int64
	closed bool
}

func NewRangeReadSeekCloser(uri string) (*RangeReadSeekCloser, error) {
	parsedUrl, err := url.Parse(uri)
	if err != nil {
		return nil, err
	}
	r := &RangeReadSeekCloser{
		uri: parsedUrl,
		buf: make([]byte, 0, rangeReaderBufferSize),
	}

	if err = r.getInformation(); err != nil {
		return nil, err
	}

	return r, nil
}

func (r *RangeReadSeekCloser) GetURI() string {
	return r.uri.String()
}

func (r *RangeReadSeekCloser) retryConnect() error {
	if r.closed {
		return errors.New("already closed")
	}

	headers := make(http.Header)
	startOffset := r.i

	expectedLength := r.size - startOffset
	headers.Set("Range", fmt.Sprintf("bytes=%d-", startOffset))
	response, err := http.DefaultClient.Do(&http.Request{
		Method: "GET",
		URL:    r.uri,
		Header: headers,
	})

	if err != nil {
		return err
	}
	r.body = response.Body
	runtime.SetFinalizer(r.body, io.ReadCloser.Close)

	if response.StatusCode != http.StatusPartialContent {
		r.body.Close()
		r.body = nil
		return fmt.Errorf("response status code %d != %d", response.StatusCode, http.StatusPartialContent)
	}

	contentLength, err := strconv.ParseInt(response.Header.Get("content-length"), 10, 0)
	if err != nil {
		r.body.Close()
		r.body = nil
		return errors.New("server response does not have a valid Content-Length")
	}

	if contentLength != expectedLength {
		r.body.Close()
		r.body = nil
		return fmt.Errorf("server returned %d bytes, expected %d", contentLength, expectedLength)
	}

	return nil
}

func (r *RangeReadSeekCloser) Read(p []byte) (n int, err error) {
	for retry := 0; retry < 2; retry++ {
		if r.i >= r.size {
			return 0, io.EOF
		}

		if r.i >= r.ib {
			bufStart := int(r.i - r.ib)
			bufEnd := bufStart + len(p)
			if bufEnd <= len(r.buf) {
				copy(p, r.buf[bufStart:bufEnd])
				r.i += int64(len(p))
				return len(p), nil
			}
		}

		if r.body == nil {
			if err = r.retryConnect(); err != nil {
				continue
			}
		}

		data := make([]byte, rangeReaderBufferSize)
		n, err = r.body.Read(data)
		if err != nil { //detect actual eof, not a disconnection
			if err == io.EOF || err == io.ErrUnexpectedEOF {
				if r.i+int64(n) != r.size {
					continue
				}
			} else {
				continue
			}
		}
		r.buf = data[:n]
		r.ib = r.i

		readBytes := len(p)
		if n < readBytes {
			readBytes = n
		}

		copy(p[:readBytes], data[:readBytes])
		r.i += int64(readBytes)
		return readBytes, nil
	}

	return 0, errors.New("could not retry")
}

func (r *RangeReadSeekCloser) Seek(offset int64, whence int) (int64, error) {
	switch whence {
	case io.SeekStart:
		r.i = offset

	case io.SeekCurrent:
		r.i += offset

	case io.SeekEnd:
		//todo: maybe without -1?
		r.i = (r.size - 1) - offset
	default:
		return 0, fmt.Errorf("unknown whence %d", whence)
	}

	if r.i >= r.size {
		return r.i, io.EOF
	} else if r.i < 0 {
		return r.i, io.ErrUnexpectedEOF
	}

	if r.i < r.ib || r.i >= r.ib+int64(len(r.buf)) {
		if r.body != nil {
			r.body.Close()
		}
		r.body = nil
	}

	return r.i, nil
}

func (r *RangeReadSeekCloser) Close() error {
	if r.body != nil {
		r.closed = true
		defer func() {
			r.body = nil
		}()
		return r.body.Close()
	}
	return nil
}

func (r *RangeReadSeekCloser) getInformation() error {
	response, err := http.DefaultClient.Head(r.GetURI())
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return fmt.Errorf("response status code %d != %d", response.StatusCode, http.StatusOK)
	}
	if response.Header.Get("accept-ranges") != "bytes" {
		return errors.New("server does not accept Range")
	}
	if r.size, err = strconv.ParseInt(response.Header.Get("content-length"), 10, 0); err != nil {
		return errors.New("server response does not have a valid Content-Length")
	}

	return nil
}