hpz937/zurg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Readd downloads mount

Browse Source
This commit is contained in:
Ben Sarmiento
2024-01-26 22:13:36 +01:00
parent ef3be36932
commit 17ab115747
9 changed files with 98 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
pkg/http/client.go
View File

@@ -28,7 +28,7 @@ type HTTPClient struct {
maxRetries int
timeoutSecs int
backoff func(attempt int) time.Duration
getRetryIncr func(resp *http.Response, reqHasRangeHeader bool, err error) int
shouldRetry func(resp *http.Response, reqHasRangeHeader bool, err error, rateLimitSleep int) int
bearerToken string
ensureIPv6Host bool
cfg config.ConfigInterface
@@ -53,58 +53,12 @@ func (e *ApiErrorResponse) Error() string {
func NewHTTPClient(token string, maxRetries int, timeoutSecs int, ensureIPv6Host bool, cfg config.ConfigInterface, log *logutil.Logger) *HTTPClient {
client := HTTPClient{
bearerToken: token,
client: &http.Client{},
maxRetries: maxRetries,
timeoutSecs: timeoutSecs,
backoff: func(attempt int) time.Duration {
maxDuration := 60
backoff := int(math.Pow(2, float64(attempt)))
if backoff > maxDuration {
backoff = maxDuration
}
return time.Duration(backoff) * time.Second
},
getRetryIncr: func(resp *http.Response, reqHasRangeHeader bool, err error) int {
if err != nil && strings.HasPrefix(err.Error(), "api response error:") {
if apiErr, ok := err.(*ApiErrorResponse); ok {
switch apiErr.Code {
case -1: // Internal error
return 1
case 5: // Slow down (retry infinitely)
time.Sleep(time.Duration(cfg.GetRateLimitSleepSeconds()) * time.Second)
return -1
case 6: // Ressource unreachable
return 1
case 17: // Hoster in maintenance
return 1
case 19: // Hoster temporarily unavailable
return 1
case 25: // Service unavailable
return 1
case 34: // Too many requests (retry infinitely)
time.Sleep(time.Duration(cfg.GetRateLimitSleepSeconds()) * time.Second)
return -1
case 36: // Fair Usage Limit
return 1
default:
return 0 // don't retry
}
}
}
if resp != nil {
if resp.StatusCode == 429 {
time.Sleep(time.Duration(cfg.GetRateLimitSleepSeconds()) * time.Second)
return -1
}
if resp.Header.Get("Content-Range") == "" && reqHasRangeHeader {
time.Sleep(10 * time.Millisecond)
return -1
}
return 0 // don't retry
}
return 1
},
bearerToken: token,
client: &http.Client{},
maxRetries: maxRetries,
timeoutSecs: timeoutSecs,
backoff: backoffFunc,
shouldRetry: shouldRetryFunc,
ensureIPv6Host: ensureIPv6Host,
cfg: cfg,
ipv6: cmap.New[string](),
@@ -120,7 +74,7 @@ func NewHTTPClient(token string, maxRetries int, timeoutSecs int, ensureIPv6Host
log.Errorf("Failed to parse proxy URL: %v", err)
return nil
}
dialer, err = proxyDialer(proxyURL)
dialer, err = client.proxyDialer(proxyURL)
if err != nil {
log.Errorf("Failed to create proxy dialer: %v", err)
return nil
@@ -170,8 +124,9 @@ func NewHTTPClient(token string, maxRetries int, timeoutSecs int, ensureIPv6Host
}
} else {
client.client.Transport = &http.Transport{
MaxIdleConns: maxConnections,
MaxConnsPerHost: maxConnections,
ResponseHeaderTimeout: time.Duration(timeoutSecs) * time.Second,
MaxIdleConns: 0,
MaxConnsPerHost: maxConnections,
DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
return dialer.Dial(network, address)
},
@@ -186,16 +141,15 @@ func (r *HTTPClient) Do(req *http.Request) (*http.Response, error) {
req.Header.Set("Authorization", "Bearer "+r.bearerToken)
}
// check if Range header is set
reqHasRangeHeader := req.Header.Get("Range") != "" && req.Header.Get("Range") != "bytes=0-"
reqHasRangeHeader := req.Header.Get("Range") != "" && !strings.HasPrefix(req.Header.Get("Range"), "bytes=0-")
var resp *http.Response
var err error
attempt := 0
for {
r.replaceHostIfNeeded(req)
r.replaceHostIfNeeded(req) // needed for ipv6
resp, err = r.client.Do(req)
if resp != nil && (resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent) {
if resp != nil && resp.StatusCode/100 >= 4 {
body, _ := io.ReadAll(resp.Body)
if body != nil {
var errResp ApiErrorResponse
@@ -206,59 +160,110 @@ func (r *HTTPClient) Do(req *http.Request) (*http.Response, error) {
}
}
}
incr := r.getRetryIncr(resp, reqHasRangeHeader, err)
incr := r.shouldRetry(resp, reqHasRangeHeader, err, r.cfg.GetRateLimitSleepSeconds())
if incr > 0 {
attempt += incr
if attempt > r.maxRetries {
break
}
if incr > 0 {
time.Sleep(r.backoff(attempt))
}
time.Sleep(r.backoff(attempt))
} else if incr == 0 {
time.Sleep(10 * time.Millisecond)
} else {
// don't retry anymore
break
}
// if incr < 0, retry infinitely
}
return resp, err
}
func (r *HTTPClient) replaceHostIfNeeded(req *http.Request) {
if !r.ensureIPv6Host && !r.cfg.ShouldForceIPv6() || !strings.HasSuffix(req.URL.Host, "download.real-debrid.com") {
if !r.ensureIPv6Host && !r.cfg.ShouldForceIPv6() || !strings.HasSuffix(req.Host, "download.real-debrid.com") {
return
}
// get subdomain of req.URL.Host
subdomain := strings.Split(req.URL.Host, ".")[0]
// get subdomain of req.Host
subdomain := strings.Split(req.Host, ".")[0]
// check if subdomain is numeric
_, err := strconv.Atoi(subdomain)
if err == nil {
// subdomain is numeric, replace it with .cloud
req.URL.Host = strings.Replace(req.URL.Host, ".com", ".cloud", 1)
req.Host = strings.Replace(req.Host, ".com", ".cloud", 1)
req.URL.Host = req.Host
}
// check if host is in the list of IPv6 hosts
found := false
for _, h := range r.ipv6Hosts {
if h == req.URL.Host {
if h == req.Host {
found = true
break
}
}
if !found {
r.log.Debug("Not found, assigning random IPv6 host")
// random IPv6 host
req.URL.Host = r.ipv6Hosts[rand.Intn(len(r.ipv6Hosts))]
req.Host = r.ipv6Hosts[rand.Intn(len(r.ipv6Hosts))]
req.URL.Host = req.Host
}
fmt.Println(req.URL.Host)
}
func proxyDialer(proxyURL *url.URL) (proxy.Dialer, error) {
func (r *HTTPClient) proxyDialer(proxyURL *url.URL) (proxy.Dialer, error) {
if proxyURL.Scheme == "http" || proxyURL.Scheme == "https" {
// Create a new HTTP proxy dialer
httpProxyDialer := http_dialer.New(proxyURL)
httpProxyDialer := http_dialer.New(proxyURL, http_dialer.WithConnectionTimeout(time.Duration(r.timeoutSecs)*time.Second))
return httpProxyDialer, nil
} else if proxyURL.Scheme == "socks5" {
// For SOCKS5 proxies, use the proxy package's FromURL
return proxy.FromURL(proxyURL, proxy.Direct)
}
return nil, fmt.Errorf("unsupported proxy scheme: %s", proxyURL.Scheme)
}
func shouldRetryFunc(resp *http.Response, reqHasRangeHeader bool, err error, rateLimitSleep int) int {
if err != nil && strings.HasPrefix(err.Error(), "api response error:") {
if apiErr, ok := err.(*ApiErrorResponse); ok {
switch apiErr.Code {
case -1: // Internal error
return 1
case 5: // Slow down (retry infinitely)
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
return 0
case 6: // Ressource unreachable
return 1
case 17: // Hoster in maintenance
return 1
case 18: // Hoster limit reached
return 1
case 19: // Hoster temporarily unavailable
return 1
case 25: // Service unavailable
return 1
case 34: // Too many requests (retry infinitely)
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
return 0
case 36: // Fair Usage Limit
return 1
default:
return -1 // don't retry
}
}
}
if resp != nil {
if resp.StatusCode == 429 {
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
return 0
}
if resp.StatusCode/100 == 2 && resp.Header.Get("Content-Range") == "" && reqHasRangeHeader {
time.Sleep(10 * time.Millisecond)
return 0
}
return -1 // don't retry
}
return 1
}
func backoffFunc(attempt int) time.Duration {
maxDuration := 60
backoff := int(math.Pow(2, float64(attempt)))
if backoff > maxDuration {
backoff = maxDuration
}
return time.Duration(backoff) * time.Second
}