mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2026-10-10 04:53:09 +08:00
893457cd50
* fix(aliyundrive): limit callback concurrency - Share proxy callback admission by Aliyun user identity and hold permits for complete response-body lifetimes. - Retry only verified callback-capacity rejections while preserving direct redirects and server download limiting. - Map exhausted temporary capacity to S3 SlowDown through the merged OpenListTeam gofakes3 module. - Cover shared limits, lifecycle release, cancellation, retry classification, and the S3 HTTP response. Co-authored-by: Codex <267193182+codex@users.noreply.github.com> # Conflicts: # go.mod # go.sum # server/s3/pager.go * fix(op): separate redirect and proxy link cache entries - Include redirect mode in the link cache key for all drivers. - Cover both redirect-to-proxy and proxy-to-redirect cache reuse. Co-authored-by: Codex <267193182+codex@users.noreply.github.com> * fix(proxy): close range bodies before opening next - make ServeHTTP own each range body and preserve cleanup failures - remove the aggregate range closer and pass range readers directly - replace the obsolete callback transport test with focused lifecycle coverage Co-authored-by: Codex <267193182+codex@users.noreply.github.com> --------- Co-authored-by: nostalume <nostalucent@gmail.com> Co-authored-by: Codex <267193182+codex@users.noreply.github.com>
156 lines
3.8 KiB
Go
156 lines
3.8 KiB
Go
package common
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
|
|
"maps"
|
|
|
|
"github.com/OpenListTeam/OpenList/v4/internal/conf"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/model"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/net"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/sign"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/stream"
|
|
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
|
|
)
|
|
|
|
func Proxy(w http.ResponseWriter, r *http.Request, link *model.Link, file model.Obj) error {
|
|
// if link.MFile != nil {
|
|
// attachHeader(w, file, link)
|
|
// http.ServeContent(w, r, file.GetName(), file.ModTime(), link.MFile)
|
|
// return nil
|
|
// }
|
|
|
|
if link.Concurrency > 0 || link.PartSize > 0 {
|
|
attachHeader(w, file, link)
|
|
size := link.ContentLength
|
|
if size <= 0 {
|
|
size = file.GetSize()
|
|
}
|
|
rrf, _ := stream.GetRangeReaderFromLink(size, link)
|
|
if link.RangeReader == nil {
|
|
r = r.WithContext(context.WithValue(r.Context(), conf.RequestHeaderKey, r.Header))
|
|
}
|
|
return net.ServeHTTP(w, r, file.GetName(), file.ModTime(), size, rrf)
|
|
}
|
|
|
|
if link.RangeReader != nil {
|
|
attachHeader(w, file, link)
|
|
size := link.ContentLength
|
|
if size <= 0 {
|
|
size = file.GetSize()
|
|
}
|
|
return net.ServeHTTP(w, r, file.GetName(), file.ModTime(), size, link.RangeReader)
|
|
}
|
|
|
|
//transparent proxy
|
|
header := net.ProcessHeader(r.Header, link.Header)
|
|
res, err := net.RequestHttp(r.Context(), r.Method, header, link.URL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer res.Body.Close()
|
|
|
|
maps.Copy(w.Header(), res.Header)
|
|
w.Header().Set("Content-Disposition", utils.GenerateContentDisposition(file.GetName()))
|
|
w.WriteHeader(res.StatusCode)
|
|
if r.Method == http.MethodHead {
|
|
return nil
|
|
}
|
|
_, err = utils.CopyWithBuffer(w, &stream.RateLimitReader{
|
|
Reader: res.Body,
|
|
Limiter: stream.ServerDownloadLimit,
|
|
Ctx: r.Context(),
|
|
})
|
|
return err
|
|
}
|
|
func attachHeader(w http.ResponseWriter, file model.Obj, link *model.Link) {
|
|
fileName := file.GetName()
|
|
w.Header().Set("Content-Disposition", utils.GenerateContentDisposition(fileName))
|
|
w.Header().Set("Content-Type", utils.GetMimeType(fileName))
|
|
size := link.ContentLength
|
|
if size <= 0 {
|
|
size = file.GetSize()
|
|
}
|
|
w.Header().Set("Etag", GetEtag(file, size))
|
|
contentType := link.Header.Get("Content-Type")
|
|
if len(contentType) > 0 {
|
|
w.Header().Set("Content-Type", contentType)
|
|
} else {
|
|
w.Header().Set("Content-Type", utils.GetMimeType(fileName))
|
|
}
|
|
}
|
|
func GetEtag(file model.Obj, size int64) string {
|
|
hash := ""
|
|
for _, v := range file.GetHash().Export() {
|
|
if v > hash {
|
|
hash = v
|
|
}
|
|
}
|
|
if len(hash) > 0 {
|
|
return fmt.Sprintf(`"%s"`, hash)
|
|
}
|
|
// 参考nginx
|
|
return fmt.Sprintf(`"%x-%x"`, file.ModTime().Unix(), size)
|
|
}
|
|
|
|
func ProxyRange(ctx context.Context, link *model.Link, size int64) *model.Link {
|
|
if link.RangeReader == nil && !strings.HasPrefix(link.URL, GetApiUrl(ctx)+"/") {
|
|
if link.ContentLength > 0 {
|
|
size = link.ContentLength
|
|
}
|
|
rrf, err := stream.GetRangeReaderFromLink(size, link)
|
|
if err == nil {
|
|
return &model.Link{
|
|
RangeReader: rrf,
|
|
ContentLength: size,
|
|
}
|
|
}
|
|
}
|
|
return link
|
|
}
|
|
|
|
type InterceptResponseWriter struct {
|
|
http.ResponseWriter
|
|
io.Writer
|
|
}
|
|
|
|
func (iw *InterceptResponseWriter) Write(p []byte) (int, error) {
|
|
return iw.Writer.Write(p)
|
|
}
|
|
|
|
type WrittenResponseWriter struct {
|
|
http.ResponseWriter
|
|
written bool
|
|
}
|
|
|
|
func (ww *WrittenResponseWriter) Write(p []byte) (int, error) {
|
|
n, err := ww.ResponseWriter.Write(p)
|
|
if !ww.written && n > 0 {
|
|
ww.written = true
|
|
}
|
|
return n, err
|
|
}
|
|
|
|
func (ww *WrittenResponseWriter) IsWritten() bool {
|
|
return ww.written
|
|
}
|
|
|
|
func GenerateDownProxyURL(storage *model.Storage, reqPath string) string {
|
|
if storage.DownProxyURL == "" {
|
|
return ""
|
|
}
|
|
query := ""
|
|
if !storage.DisableProxySign {
|
|
query = "?sign=" + sign.Sign(reqPath)
|
|
}
|
|
return fmt.Sprintf("%s%s%s",
|
|
strings.Split(storage.DownProxyURL, "\n")[0],
|
|
utils.EncodePath(reqPath, true),
|
|
query,
|
|
)
|
|
}
|