diff --git a/server/handlers/proxy/cache.go b/server/handlers/proxy/cache.go index e8247d6..0f78b83 100644 --- a/server/handlers/proxy/cache.go +++ b/server/handlers/proxy/cache.go @@ -26,9 +26,10 @@ type Cache interface { // CacheMetadata stores metadata about a cached response type CacheMetadata struct { - Headers http.Header `json:"headers"` - ContentType string `json:"content_type"` - ContentTotalLength int64 `json:"content_total_length"` + Headers http.Header `json:"headers,omitempty"` + ContentType string `json:"content_type,omitempty"` + ContentTotalLength int64 `json:"content_total_length,omitempty"` + NotSupportRange bool `json:"not_support_range,omitempty"` } func (m *CacheMetadata) MarshalBinary() ([]byte, error) { diff --git a/server/handlers/proxy/proxy.go b/server/handlers/proxy/proxy.go index fb33e4a..cf1d41b 100644 --- a/server/handlers/proxy/proxy.go +++ b/server/handlers/proxy/proxy.go @@ -130,7 +130,6 @@ func ProxyURL(ctx *gin.Context, u string, headers map[string]string, opts ...Pro rsc := NewHttpReadSeekCloser(u, WithContext(c), WithHeadersMap(headers), - WithNotSupportRange(ctx.GetHeader("Range") == ""), ) defer rsc.Close() if o.CacheKey == "" { diff --git a/server/handlers/proxy/readseeker.go b/server/handlers/proxy/readseeker.go index 5470f6b..7551ca2 100644 --- a/server/handlers/proxy/readseeker.go +++ b/server/handlers/proxy/readseeker.go @@ -32,9 +32,11 @@ type HttpReadSeekCloser struct { allowedStatusCodes []int offset int64 contentTotalLength int64 - length int64 + perLength int64 currentRespMaxOffset int64 notSupportRange bool + // if the server does not support range requests, the seek method will be unusable + notSupportSeekWhenNotSupportRange bool } type HttpReadSeekerConf func(h *HttpReadSeekCloser) @@ -119,27 +121,34 @@ func NotAllowedStatusCodes(codes ...int) HttpReadSeekerConf { } } -func WithLength(length int64) HttpReadSeekerConf { +// sets the per length of the request +func WithPerLength(length int64) HttpReadSeekerConf { return func(h *HttpReadSeekCloser) { if length > 0 { - h.length = length + h.perLength = length } } } -func WithNotSupportRange(notSupportRange bool) HttpReadSeekerConf { +func WithForceNotSupportRange(notSupportRange bool) HttpReadSeekerConf { return func(h *HttpReadSeekCloser) { h.notSupportRange = notSupportRange } } +func WithNotSupportSeekWhenNotSupportRange(notSupportSeekWhenNotSupportRange bool) HttpReadSeekerConf { + return func(h *HttpReadSeekCloser) { + h.notSupportSeekWhenNotSupportRange = notSupportSeekWhenNotSupportRange + } +} + func NewHttpReadSeekCloser(url string, conf ...HttpReadSeekerConf) *HttpReadSeekCloser { rs := &HttpReadSeekCloser{ url: url, contentTotalLength: -1, method: http.MethodGet, headMethod: http.MethodHead, - length: 1024 * 1024 * 16, + perLength: 1024 * 1024 * 16, headers: make(http.Header), client: http.DefaultClient, } @@ -171,8 +180,8 @@ func (h *HttpReadSeekCloser) fix() *HttpReadSeekCloser { if len(h.notAllowedStatusCodes) == 0 { h.notAllowedStatusCodes = []int{http.StatusNotFound} } - if h.length <= 0 { - h.length = 64 * 1024 + if h.perLength <= 0 { + h.perLength = 1024 * 1024 } if h.headers == nil { h.headers = make(http.Header) @@ -229,22 +238,36 @@ func (h *HttpReadSeekCloser) FetchNextChunk() error { resp, err := h.client.Do(req) if err != nil { - return fmt.Errorf("failed to execute HTTP request: %w", err) + return fmt.Errorf("failed to execute request: %w", err) } h.contentType = resp.Header.Get("Content-Type") if resp.StatusCode == http.StatusOK { + if ar := resp.Header.Get("Accept-Ranges"); ar == "" || ar == "none" { + h.notSupportRange = true + } + // if the maximum offset of the current response is less than the content length minus one, it means that the server does not support range requests - if h.currentRespMaxOffset < resp.ContentLength-1 || h.notSupportRange { - // if the offset is not 0, it means that the seek method is incorrectly used - if h.offset != 0 { - resp.Body.Close() - return fmt.Errorf("server does not support range requests, cannot seek to non-zero offset") - } + if h.currentRespMaxOffset < resp.ContentLength-1 { h.notSupportRange = true + } + + if h.notSupportRange { h.contentTotalLength = resp.ContentLength h.currentRespMaxOffset = h.contentTotalLength - 1 + + // If offset > 0, read and discard bytes until reaching the desired offset + if h.offset > 0 { + if _, err := io.CopyN(io.Discard, resp.Body, h.offset); err != nil { + resp.Body.Close() + if err == io.EOF { + return io.EOF + } + return fmt.Errorf("failed to discard bytes: %w", err) + } + } + h.currentResp = resp return nil } @@ -254,12 +277,22 @@ func (h *HttpReadSeekCloser) FetchNextChunk() error { resp.Body.Close() return h.FetchNextChunk() } - // if the offset is greater than 0, it means that the seek method is incorrectly used - if h.offset > 0 { + + if h.contentTotalLength != resp.ContentLength { resp.Body.Close() - return fmt.Errorf("server does not support range requests, cannot seek to offset %d", h.offset) + return fmt.Errorf("content length mismatch: %d != %d", h.contentTotalLength, resp.ContentLength) } + h.notSupportRange = true + if h.offset > 0 { + if _, err := io.CopyN(io.Discard, resp.Body, h.offset); err != nil { + resp.Body.Close() + if err == io.EOF { + return io.EOF + } + return fmt.Errorf("failed to discard bytes: %w", err) + } + } h.currentRespMaxOffset = h.contentTotalLength - 1 h.currentResp = resp return nil @@ -267,7 +300,7 @@ func (h *HttpReadSeekCloser) FetchNextChunk() error { if resp.StatusCode != http.StatusPartialContent { resp.Body.Close() - return fmt.Errorf("unexpected HTTP status code: %d (expected 206 Partial Content)", resp.StatusCode) + return fmt.Errorf("unexpected status code: %d", resp.StatusCode) } if err := h.checkResponse(resp); err != nil { @@ -301,7 +334,7 @@ func (h *HttpReadSeekCloser) createRequest() (*http.Request, error) { return nil, err } - end := h.offset + h.length - 1 + end := h.offset + h.perLength - 1 if h.contentTotalLength > 0 && end > h.contentTotalLength-1 { end = h.contentTotalLength - 1 } @@ -315,7 +348,7 @@ func (h *HttpReadSeekCloser) createRequest() (*http.Request, error) { func (h *HttpReadSeekCloser) createRequestWithoutRange() (*http.Request, error) { req, err := http.NewRequestWithContext(h.ctx, h.method, h.url, nil) if err != nil { - return nil, fmt.Errorf("failed to create HTTP request: %w", err) + return nil, fmt.Errorf("failed to create request: %w", err) } req.Header = h.headers.Clone() req.Header.Del("Range") @@ -351,13 +384,13 @@ func (h *HttpReadSeekCloser) checkContentType(ct string) error { func (h *HttpReadSeekCloser) checkStatusCode(code int) error { if len(h.allowedStatusCodes) != 0 { if slices.Index(h.allowedStatusCodes, code) == -1 { - return fmt.Errorf("HTTP status code %d is not in the list of allowed status codes: %v", code, h.allowedStatusCodes) + return fmt.Errorf("status code %d is not in the list of allowed status codes: %v", code, h.allowedStatusCodes) } return nil } if len(h.notAllowedStatusCodes) != 0 { if slices.Index(h.notAllowedStatusCodes, code) != -1 { - return fmt.Errorf("HTTP status code %d is in the list of not allowed status codes: %v", code, h.notAllowedStatusCodes) + return fmt.Errorf("status code %d is in the list of not allowed status codes: %v", code, h.notAllowedStatusCodes) } } return nil @@ -366,15 +399,20 @@ func (h *HttpReadSeekCloser) checkStatusCode(code int) error { func (h *HttpReadSeekCloser) Seek(offset int64, whence int) (int64, error) { newOffset, err := h.calculateNewOffset(offset, whence) if err != nil { + h.closeCurrentResp() return 0, fmt.Errorf("failed to calculate new offset: %w", err) } if newOffset < 0 { + h.closeCurrentResp() return 0, fmt.Errorf("cannot seek to negative offset: %d", newOffset) } if newOffset != h.offset { h.closeCurrentResp() + if h.notSupportRange && h.notSupportSeekWhenNotSupportRange { + return 0, fmt.Errorf("seek is not supported when not support range") + } h.offset = newOffset } @@ -384,14 +422,8 @@ func (h *HttpReadSeekCloser) Seek(offset int64, whence int) (int64, error) { func (h *HttpReadSeekCloser) calculateNewOffset(offset int64, whence int) (int64, error) { switch whence { case io.SeekStart: - if h.notSupportRange && offset != 0 && offset != h.offset { - return 0, fmt.Errorf("server does not support range requests, cannot seek to non-zero offset") - } return offset, nil case io.SeekCurrent: - if h.notSupportRange && offset != 0 { - return 0, fmt.Errorf("server does not support range requests, cannot seek to non-zero offset") - } return h.offset + offset, nil case io.SeekEnd: if h.contentTotalLength < 0 { @@ -399,13 +431,9 @@ func (h *HttpReadSeekCloser) calculateNewOffset(offset int64, whence int) (int64 return 0, fmt.Errorf("failed to fetch content length: %w", err) } } - newOffset := h.contentTotalLength - offset - if h.notSupportRange && newOffset != h.offset { - return 0, fmt.Errorf("server does not support range requests, cannot seek to non-zero offset") - } - return newOffset, nil + return h.contentTotalLength - offset, nil default: - return 0, fmt.Errorf("invalid seek whence value: %d (must be 0, 1, or 2)", whence) + return 0, fmt.Errorf("invalid seek whence value: %d", whence) } } @@ -463,6 +491,10 @@ func (h *HttpReadSeekCloser) ContentType() (string, error) { return "", fmt.Errorf("content type is not available - no successful response received yet") } +func (h *HttpReadSeekCloser) AcceptRanges() bool { + return !h.notSupportRange +} + func (h *HttpReadSeekCloser) ContentTotalLength() (int64, error) { if h.contentTotalLength > 0 { return h.contentTotalLength, nil diff --git a/server/handlers/proxy/slice.go b/server/handlers/proxy/slice.go index da88185..dfbbf30 100644 --- a/server/handlers/proxy/slice.go +++ b/server/handlers/proxy/slice.go @@ -18,6 +18,7 @@ var mu = ksync.DefaultKmutex() // Proxy defines the interface for proxy implementations type Proxy interface { io.ReadSeeker + AcceptRanges() bool ContentTotalLength() (int64, error) ContentType() (string, error) } @@ -133,7 +134,7 @@ func (c *SliceCacheProxy) Proxy(w http.ResponseWriter, r *http.Request) error { cacheItem, err := c.getCacheItem(alignedOffset) if err != nil { http.Error(w, fmt.Sprintf("Failed to get cache item: %v", err), http.StatusInternalServerError) - return err + return fmt.Errorf("failed to get cache item: %w", err) } c.setResponseHeaders(w, byteRange, cacheItem, isRangeRequest) @@ -154,10 +155,14 @@ func (c *SliceCacheProxy) setResponseHeaders(w http.ResponseWriter, byteRange *B } } + if !cacheItem.Metadata.NotSupportRange { + w.Header().Set("Accept-Ranges", "bytes") + } else { + w.Header().Set("Accept-Ranges", "none") + } w.Header().Set("Content-Length", fmtContentLength(byteRange.Start, byteRange.End, cacheItem.Metadata.ContentTotalLength)) w.Header().Set("Content-Type", cacheItem.Metadata.ContentType) if isRangeRequest { - w.Header().Set("Accept-Ranges", "bytes") w.Header().Set("Content-Range", fmtContentRange(byteRange.Start, byteRange.End, cacheItem.Metadata.ContentTotalLength)) w.WriteHeader(http.StatusPartialContent) } else { @@ -246,6 +251,17 @@ func (c *SliceCacheProxy) getCacheItem(alignedOffset int64) (*CacheItem, error) return slice, nil } +func (c *SliceCacheProxy) contentTotalLength() (int64, error) { + total, err := c.r.ContentTotalLength() + if err != nil { + return -1, fmt.Errorf("failed to get content total length from source: %w", err) + } + if total == -1 { + return -1, fmt.Errorf("source does not support range requests") + } + return total, nil +} + func (c *SliceCacheProxy) fetchFromSource(offset int64) (*CacheItem, error) { if offset < 0 { return nil, fmt.Errorf("source offset cannot be negative, got: %d", offset) @@ -254,22 +270,28 @@ func (c *SliceCacheProxy) fetchFromSource(offset int64) (*CacheItem, error) { return nil, fmt.Errorf("failed to seek to offset %d in source: %w", offset, err) } + var total int64 = -1 + buf := make([]byte, c.sliceSize) n, err := io.ReadFull(c.r, buf) - if err != nil && err != io.ErrUnexpectedEOF { - return nil, fmt.Errorf("failed to read %d bytes from source at offset %d: %w", c.sliceSize, offset, err) - } - - var headers http.Header - if h, ok := c.r.(Headers); ok { - headers = h.Headers().Clone() - } else { - headers = make(http.Header) + if err != nil { + if err != io.ErrUnexpectedEOF { + return nil, fmt.Errorf("failed to read %d bytes from source at offset %d: %w", c.sliceSize, offset, err) + } + total, err = c.contentTotalLength() + if err != nil { + return nil, fmt.Errorf("failed to get content total length from source: %w", err) + } + if total != offset+int64(n) { + return nil, fmt.Errorf("source content total length mismatch, got: %d, expected: %d", total, offset+int64(n)) + } } - contentTotalLength, err := c.r.ContentTotalLength() - if err != nil { - return nil, fmt.Errorf("failed to get content total length from source: %w", err) + if total == -1 { + total, err = c.contentTotalLength() + if err != nil { + return nil, fmt.Errorf("failed to get content total length from source: %w", err) + } } contentType, err := c.r.ContentType() @@ -277,11 +299,19 @@ func (c *SliceCacheProxy) fetchFromSource(offset int64) (*CacheItem, error) { return nil, fmt.Errorf("failed to get content type from source: %w", err) } + var headers http.Header + if h, ok := c.r.(Headers); ok { + headers = h.Headers().Clone() + } else { + headers = make(http.Header) + } + return &CacheItem{ Metadata: &CacheMetadata{ Headers: headers, - ContentTotalLength: contentTotalLength, + ContentTotalLength: total, ContentType: contentType, + NotSupportRange: !c.r.AcceptRanges(), }, Data: buf[:n], }, nil @@ -305,6 +335,10 @@ func ParseByteRange(r string) (*ByteRange, error) { return nil, fmt.Errorf("range header must start with 'bytes=', got: %s", r) } + if strings.Contains(r, ",") { + return nil, fmt.Errorf("not support multi-range, got: %s", r) + } + r = strings.TrimPrefix(r, "bytes=") parts := strings.Split(r, "-") if len(parts) != 2 {