mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-13 20:03:46 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Levi
|
44e2272ca4 |
No files matched your search
@@ -798,5 +798,5 @@ jobs:
|
||||
linux/arm64
|
||||
linux/arm/v7
|
||||
tags: |
|
||||
${{ !contains(env.VERSION, 'beta') && 'liwei2633/gopeed:latest' || '' }}
|
||||
liwei2633/gopeed:latest
|
||||
liwei2633/gopeed:v${{ env.VERSION }}
|
||||
@@ -86,32 +86,6 @@ jobs:
|
||||
flutter pub get
|
||||
flutter gen-l10n
|
||||
flutter test test/api/downloader_config_test.dart
|
||||
test-flutter-task-context-menu:
|
||||
name: Test task context menus and text inputs
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 20
|
||||
env:
|
||||
CHROME_EXECUTABLE: /usr/bin/google-chrome
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: subosito/flutter-action@v2
|
||||
with:
|
||||
flutter-version: ${{ env.FLUTTER_VERSION }}
|
||||
- name: Prepare Flutter tests
|
||||
working-directory: ui/flutter
|
||||
run: |
|
||||
flutter pub get
|
||||
flutter gen-l10n
|
||||
- name: Test task menu actions and native input policy
|
||||
working-directory: ui/flutter
|
||||
run: |
|
||||
flutter test --no-pub test/features/tasks/presentation/widgets/task_context_menu_test.dart test/shared/widgets/app_text_field_test.dart
|
||||
flutter test --no-pub test/widget_test.dart --plain-name 'task context menu exposes'
|
||||
- name: Test browser menu cancellation and input policy
|
||||
working-directory: ui/flutter
|
||||
run: |
|
||||
"$CHROME_EXECUTABLE" --version
|
||||
flutter test --no-pub --platform chrome test/features/tasks/presentation/widgets/task_context_menu_test.dart test/shared/widgets/app_text_field_test.dart
|
||||
test-webview-integration:
|
||||
name: Test webview integration (${{ matrix.name }})
|
||||
strategy:
|
||||
|
||||
@@ -20,7 +20,7 @@ Beyond core download management, Gopeed offers browser integration, JavaScript e
|
||||
|
||||
Visit ✈ [Official Website](https://gopeed.com)
|
||||
|
||||
|
||||
|
||||
|
||||
## ✨ Features
|
||||
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@ Gopeed（**Go Speed** の略）は、Go と Flutter で開発された高速で
|
||||
|
||||
✈ [公式サイト](https://gopeed.com)
|
||||
|
||||
|
||||
|
||||
|
||||
## ✨ 主な機能
|
||||
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@ Ngoài việc quản lý tải xuống hằng ngày, Gopeed còn hỗ trợ tíc
|
||||
|
||||
Truy cập ✈ [Trang web chính thức](https://gopeed.com)
|
||||
|
||||
|
||||
|
||||
|
||||
## ✨ Tính năng chính
|
||||
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@ Gopeed（全称 **Go Speed**，也叫“够快下载器”）是一款使用 Go
|
||||
|
||||
访问 ✈ [官方网站](https://gopeed.com)
|
||||
|
||||
|
||||
|
||||
|
||||
## ✨ 主要特性
|
||||
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@ Gopeed（**Go Speed** 的縮寫）是一款以 Go 與 Flutter 開發的高速、
|
||||
|
||||
瀏覽 ✈ [官方網站](https://gopeed.com)
|
||||
|
||||
|
||||
|
||||
|
||||
## ✨ 主要功能
|
||||
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 74 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 77 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 1.3 MiB |
Binary file not shown.
|
Before Width: | Height: | Size: 792 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 1.3 MiB |
Binary file not shown.
|
Before Width: | Height: | Size: 787 KiB |
@@ -1,52 +0,0 @@
|
||||
# Batch task creation
|
||||
|
||||
`POST /api/v1/tasks/batch` accepts ordinary and grouped batches:
|
||||
|
||||
```json
|
||||
{
|
||||
"group": true,
|
||||
"groupName": "album",
|
||||
"opts": {"path": "/downloads"},
|
||||
"reqs": [
|
||||
{"rid": "resolved-resource-id", "opts": {"selectFiles": [1]}},
|
||||
{"req": {"url": "https://example.com/cover.jpg"}}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
Each member accepts `rid` or `req`. A nonempty `rid` takes priority and creates a
|
||||
task from the cached resolve result. A member's `opts`, when present, replaces
|
||||
batch `opts` as a whole; omitted fields do not inherit individual batch values.
|
||||
Members without `opts` use batch `opts`. BT file selections use the indexes from
|
||||
the resolve result. Requests without `rid` continue to work.
|
||||
|
||||
For groups, batch `opts.path` selects the parent directory. Members inheriting
|
||||
batch options, or with an empty path, use the reserved group directory. Explicit
|
||||
relative member paths are resolved under that directory; do not prefix them with
|
||||
the batch parent path. For example, parent `downloads` and member path `nested`
|
||||
place the member in `downloads/<group name>/nested`. Absolute paths under the
|
||||
parent directory preserve their relative structure within the group. In ordinary
|
||||
batches, an empty member path uses the configured default download directory.
|
||||
|
||||
All members are attempted. If at least one task is persisted, the response is
|
||||
successful and contains the created task IDs. A task that fails to start after
|
||||
being persisted is included; use the existing task controls to retry it.
|
||||
Successful tasks are kept when other members fail. If every member fails, the
|
||||
API returns an error and removes any newly reserved empty group and directory.
|
||||
|
||||
Both ordinary and grouped batches now return an object in `data`:
|
||||
|
||||
```json
|
||||
{"code": 0, "msg": "", "data": {"taskIds": ["task-1", "task-2"], "parentId": "group-id"}}
|
||||
```
|
||||
|
||||
Ordinary batches omit `parentId`. **This replaces the former array response for
|
||||
ordinary batches.** The outer `code`/`msg`/`data` envelope is unchanged. Clients
|
||||
must read `data.taskIds`; there are no per-member results. A successful response
|
||||
completes the batch, including partial success, so clients must not automatically
|
||||
resubmit missing members.
|
||||
|
||||
With `group: true`, the server generates the group ID and reserves its directory.
|
||||
Creation does not accept a client group ID or append to an existing group. Group
|
||||
metadata remains available through the existing task list and detail responses;
|
||||
`parentId` identifies membership for group operations.
|
||||
+1
-7
@@ -1,9 +1,6 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/GopeedLab/gopeed/cmd"
|
||||
"github.com/GopeedLab/gopeed/pkg/rest/model"
|
||||
)
|
||||
@@ -16,8 +13,5 @@ func main() {
|
||||
Storage: model.StorageBolt,
|
||||
WebEnable: true,
|
||||
}
|
||||
if err := cmd.Start(cfg); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
cmd.Start(cfg)
|
||||
}
|
||||
+5
-7
@@ -16,16 +16,15 @@ import (
|
||||
//go:embed banner.txt
|
||||
var banner string
|
||||
|
||||
func Start(cfg *model.StartConfig) error {
|
||||
func Start(cfg *model.StartConfig) {
|
||||
fmt.Println(banner)
|
||||
srv, listener, err := rest.BuildServer(cfg)
|
||||
if err != nil {
|
||||
return err
|
||||
panic(err)
|
||||
}
|
||||
fmt.Println(banner)
|
||||
downloadCfg, err := rest.Downloader.GetConfig()
|
||||
if err != nil {
|
||||
listener.Close()
|
||||
return err
|
||||
panic(err)
|
||||
}
|
||||
if rest.Downloader.IsFirstLoad() {
|
||||
// Set default download config
|
||||
@@ -63,9 +62,8 @@ func Start(cfg *model.StartConfig) error {
|
||||
|
||||
fmt.Printf("Server start success on http://%s\n", listener.Addr().String())
|
||||
if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed {
|
||||
return err
|
||||
panic(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func watchExit() {
|
||||
|
||||
@@ -1,35 +0,0 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/rest/model"
|
||||
)
|
||||
|
||||
func TestStartInvalidWebAPITokenReturnsError(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
webAuth *model.WebAuth
|
||||
}{
|
||||
{name: "no credentials"},
|
||||
{name: "missing username", webAuth: &model.WebAuth{Password: "secret"}},
|
||||
{name: "missing password", webAuth: &model.WebAuth{Username: "gopeed"}},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := Start(&model.StartConfig{
|
||||
WebEnable: true,
|
||||
ApiToken: "test-token",
|
||||
WebAuth: tt.webAuth,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("Start() error = nil, want web authentication validation error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "web authentication username and password are required") {
|
||||
t.Fatalf("Start() error = %q, want missing web authentication credentials", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -52,8 +52,8 @@ func loadCliArgs() *args {
|
||||
flag.StringVar(cfg.Username, "username", "gopeed", "Web Authentication Username")
|
||||
cfg.Password = flag.String("p", "", "Web Authentication Password, if no password is set, web authentication will not be enabled")
|
||||
flag.StringVar(cfg.Password, "password", "", "Web Authentication Password, if no password is set, web authentication will not be enabled")
|
||||
cfg.ApiToken = flag.String("T", "", "API token (requires web authentication with username and password)")
|
||||
flag.StringVar(cfg.ApiToken, "api-token", "", "API token (requires web authentication with username and password)")
|
||||
cfg.ApiToken = flag.String("T", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
|
||||
flag.StringVar(cfg.ApiToken, "api-token", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
|
||||
flag.BoolVar(&cfg.MCPEnable, "mcp-enable", false, "Enable MCP endpoint")
|
||||
cfg.TempDir = flag.String("temp-dir", "", "Temporary directory (defaults to the OS temporary directory)")
|
||||
cfg.StorageDir = flag.String("d", "", "Storage directory")
|
||||
|
||||
+1
-4
@@ -62,10 +62,7 @@ func main() {
|
||||
WebFS: sub,
|
||||
WebAuth: webAuth,
|
||||
}
|
||||
if err := cmd.Start(cfg); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
cmd.Start(cfg)
|
||||
}
|
||||
|
||||
func isNotBlank(str *string) bool {
|
||||
|
||||
@@ -10,13 +10,11 @@ require (
|
||||
github.com/bodgit/sevenzip v1.6.1
|
||||
github.com/dop251/goja v0.0.0-20260106131823-651366fbe6e3
|
||||
github.com/dop251/goja_nodejs v0.0.0-20240728170619-29b559befffc
|
||||
github.com/fclairamb/ftpserverlib v0.31.0
|
||||
github.com/go-git/go-git/v5 v5.8.1
|
||||
github.com/google/jsonschema-go v0.4.3
|
||||
github.com/gorilla/handlers v1.5.1
|
||||
github.com/gorilla/mux v1.8.0
|
||||
github.com/imroc/req/v3 v3.52.2
|
||||
github.com/jlaffaye/ftp v0.2.4
|
||||
github.com/matoous/go-nanoid/v2 v2.0.0
|
||||
github.com/mattn/go-ieproxy v0.0.12
|
||||
github.com/mholt/archives v0.1.5
|
||||
@@ -25,7 +23,6 @@ require (
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529
|
||||
github.com/rs/zerolog v1.31.0
|
||||
github.com/spf13/afero v1.15.0
|
||||
github.com/tetratelabs/wazero v1.12.0
|
||||
github.com/xiaoqidun/setft v0.0.0-20220310121541-be86327699ad
|
||||
go.etcd.io/bbolt v1.4.3
|
||||
@@ -62,6 +59,7 @@ require (
|
||||
github.com/segmentio/asm v1.1.3 // indirect
|
||||
github.com/segmentio/encoding v0.5.4 // indirect
|
||||
github.com/sorairolake/lzip-go v0.3.8 // indirect
|
||||
github.com/spf13/afero v1.15.0 // indirect
|
||||
github.com/ulikunitz/xz v0.5.15 // indirect
|
||||
github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||
|
||||
@@ -205,8 +205,6 @@ github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc
|
||||
github.com/emirpasic/gods v1.18.1/go.mod h1:8tpGGwCnJ5H4r6BWwaV6OrWmMoPhUl5jm/FMNAnJvWQ=
|
||||
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/fclairamb/ftpserverlib v0.31.0 h1:ws4fLJvuU/mgg1uIgRjCJfc6CrFSgIBAOxVsnawXegE=
|
||||
github.com/fclairamb/ftpserverlib v0.31.0/go.mod h1:C5hUT0sd9P5XsHMxsmpcrT6wdZexu3qtUu1z0Ih6aG0=
|
||||
github.com/felixge/fgprof v0.9.5 h1:8+vR6yu2vvSKn08urWyEuxx75NWPEvybbkBirEpsbVY=
|
||||
github.com/felixge/fgprof v0.9.5/go.mod h1:yKl+ERSa++RYOs32d8K6WEXCB4uXdLls4ZaZPpayhMM=
|
||||
github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8SPQ=
|
||||
@@ -349,8 +347,6 @@ github.com/imroc/req/v3 v3.52.2 h1:xJocr1aIv0a2K9knfBQ4JnZHk+kWTITdjf0mgDg229I=
|
||||
github.com/imroc/req/v3 v3.52.2/go.mod h1:dBGsDloOSZJcFs6PnTjZXYBJK70OXbZpizHBLNqcH2k=
|
||||
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
|
||||
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
|
||||
github.com/jlaffaye/ftp v0.2.4 h1:JqI85DdkfZj8ntaHk8W9U2SC3jNfiPUU70+wtIWmlfE=
|
||||
github.com/jlaffaye/ftp v0.2.4/go.mod h1:Y1ZnkzxownGIuX7xQ1mQzzkZ21+DbjVIyeKL/V+IIz4=
|
||||
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
|
||||
github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
|
||||
github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
|
||||
@@ -531,8 +527,6 @@ github.com/rs/zerolog v1.31.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWR
|
||||
github.com/rwcarlsen/goexif v0.0.0-20190401172101-9e8deecbddbd/go.mod h1:hPqNNc0+uJM6H+SuU8sEs5K5IQeKccPqeSjfgcKGgPk=
|
||||
github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8=
|
||||
github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8=
|
||||
github.com/secsy/goftp v0.0.0-20200609142545-aa2de14babf4 h1:PT+ElG/UUFMfqy5HrxJxNzj3QBOf7dZwupeVC+mG1Lo=
|
||||
github.com/secsy/goftp v0.0.0-20200609142545-aa2de14babf4/go.mod h1:MnkX001NG75g3p8bhFycnyIjeQoOjGL6CEIsdE/nKSY=
|
||||
github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc=
|
||||
github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
|
||||
github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0=
|
||||
@@ -572,8 +566,6 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
|
||||
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
|
||||
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
|
||||
github.com/tetratelabs/wazero v1.12.0 h1:DuWcpNu/FzgEXgGBDp8J1Spc+CWOvvtvVyjKlaZopYU=
|
||||
github.com/tetratelabs/wazero v1.12.0/go.mod h1:LvKtzl2RqO4gyF27BiXU+nKAjcV8f38U+kP/q2vgxh0=
|
||||
github.com/tidwall/btree v1.8.1 h1:27ehoXvm5AG/g+1VxLS1SD3vRhp/H7LuEfwNvddEdmA=
|
||||
@@ -620,8 +612,6 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
go.uber.org/mock v0.5.1 h1:ASgazW/qBmR+A32MYFDB6E2POoTgOwT509VP0CT/fjs=
|
||||
go.uber.org/mock v0.5.1/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
|
||||
go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
|
||||
go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
|
||||
go4.org v0.0.0-20230225012048-214862532bf5 h1:nifaUDeh+rPaBCMPMQHZmvJf+QdpLFnuQPwx+LxVmtc=
|
||||
go4.org v0.0.0-20230225012048-214862532bf5/go.mod h1:F57wTi5Lrj6WLyswp5EYV1ncrEbFGHD4hhz6S1ZYeaU=
|
||||
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
|
||||
|
||||
@@ -3,7 +3,6 @@ package controller
|
||||
import (
|
||||
"github.com/GopeedLab/gopeed/internal/tempfiles"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
"github.com/rs/zerolog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
@@ -16,9 +15,6 @@ type Controller struct {
|
||||
TempDir string
|
||||
GetConfig func(v any)
|
||||
GetProxy func(requestProxy *base.RequestProxy) func(*http.Request) (*url.URL, error)
|
||||
// Logger is the downloader's core logger. It may be nil, in which case
|
||||
// fetchers log nothing.
|
||||
Logger *zerolog.Logger
|
||||
FileController
|
||||
//ContextDialer() (proxy.Dialer, error)
|
||||
}
|
||||
|
||||
@@ -67,11 +67,16 @@ func (f *Fetcher) Setup(ctl *controller.Controller) {
|
||||
return
|
||||
}
|
||||
|
||||
// initClient must be called with lock held.
|
||||
func (f *Fetcher) initClient() (err error) {
|
||||
lock.Lock()
|
||||
defer lock.Unlock()
|
||||
|
||||
if client != nil {
|
||||
return
|
||||
}
|
||||
if closeCtx == nil {
|
||||
closeCtx, closeFunc = context.WithCancel(context.Background())
|
||||
}
|
||||
|
||||
cfg = torrent.NewDefaultClientConfig()
|
||||
cfg.Seed = true
|
||||
@@ -89,23 +94,12 @@ func (f *Fetcher) initClient() (err error) {
|
||||
}
|
||||
|
||||
closeCtx, closeFunc = context.WithCancel(context.Background())
|
||||
// closeClient may reset closeCtx before the goroutine runs, so pass it as an argument.
|
||||
go dnsResolver.Run(closeCtx)
|
||||
go func() {
|
||||
dnsResolver.Run(closeCtx)
|
||||
}()
|
||||
return
|
||||
}
|
||||
|
||||
// addTorrentSpec holds lock from client initialization until the torrent is added,
|
||||
// so closeClient cannot close or reset the client in between.
|
||||
func (f *Fetcher) addTorrentSpec(spec *torrent.TorrentSpec) (*torrent.Torrent, bool, error) {
|
||||
lock.Lock()
|
||||
defer lock.Unlock()
|
||||
|
||||
if err := f.initClient(); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return client.AddTorrentSpec(spec)
|
||||
}
|
||||
|
||||
func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
@@ -754,6 +748,9 @@ func (f *Fetcher) addTorrent(req *base.Request) (err error) {
|
||||
if err = base.ParseReqExtra[bt.ReqExtra](req); err != nil {
|
||||
return
|
||||
}
|
||||
if err = f.initClient(); err != nil {
|
||||
return
|
||||
}
|
||||
schema := util.ParseSchema(req.URL)
|
||||
privateTorrent := false
|
||||
var spec *torrent.TorrentSpec
|
||||
@@ -832,7 +829,7 @@ func (f *Fetcher) addTorrent(req *base.Request) (err error) {
|
||||
s := &deferredFileStorage{}
|
||||
spec.Storage = s
|
||||
var added bool
|
||||
f.torrent, added, err = f.addTorrentSpec(spec)
|
||||
f.torrent, added, err = client.AddTorrentSpec(spec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/controller"
|
||||
@@ -72,82 +71,6 @@ func TestFetcher_ResolveWithProxy(t *testing.T) {
|
||||
}))
|
||||
}
|
||||
|
||||
func TestFetcher_CloseAfterInit(t *testing.T) {
|
||||
if err := closeClient(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 100; i++ {
|
||||
f := buildFetcher(t).(*Fetcher)
|
||||
f.meta.Req = &base.Request{}
|
||||
lock.Lock()
|
||||
err := f.initClient()
|
||||
lock.Unlock()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := closeClient(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetcher_CloseWithoutClient(t *testing.T) {
|
||||
if err := closeClient(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Run("restored task", func(t *testing.T) {
|
||||
_, build := new(FetcherManager).Restore()
|
||||
f := build(&fetcher.FetcherMeta{
|
||||
Req: &base.Request{URL: "./testdata/test.torrent"},
|
||||
Opts: &base.Options{},
|
||||
}, &fetcherData{})
|
||||
f.Setup(controller.NewController())
|
||||
if err := f.Close(); err != nil {
|
||||
t.Errorf("Close() got = %v, want nil", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("failed resolve", func(t *testing.T) {
|
||||
f := buildFetcher(t)
|
||||
if err := f.Resolve(&base.Request{URL: "./testdata/not-exist.torrent"}); err == nil {
|
||||
t.Fatal("Resolve() got = nil, want error")
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
t.Errorf("Close() got = %v, want nil", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestFetcher_ResolveWhileClientCloses(t *testing.T) {
|
||||
if err := closeClient(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { closeClient() })
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, u := range []string{"./testdata/test.torrent", "./testdata/ubuntu-22.04-live-server-amd64.iso.torrent"} {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 10; i++ {
|
||||
f := buildFetcher(t)
|
||||
// The other goroutine may close the client after the torrent was added.
|
||||
if err := f.Resolve(&base.Request{URL: u}); err != nil && err.Error() != "torrent closed during resolution" {
|
||||
t.Errorf("Resolve(%s) got = %v", u, err)
|
||||
return
|
||||
}
|
||||
f.Close()
|
||||
if err := closeClient(); err != nil {
|
||||
t.Errorf("closeClient() got = %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func doResolve(t *testing.T, fetcher fetcher.Fetcher) {
|
||||
t.Run("Resolve Single File", func(t *testing.T) {
|
||||
err := fetcher.Resolve(&base.Request{
|
||||
|
||||
@@ -1,404 +0,0 @@
|
||||
package ftp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/textproto"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/jlaffaye/ftp"
|
||||
"golang.org/x/net/proxy"
|
||||
|
||||
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
|
||||
)
|
||||
|
||||
const (
|
||||
connectTimeout = 15 * time.Second
|
||||
readTimeout = 15 * time.Second
|
||||
quitTimeout = 2 * time.Second
|
||||
)
|
||||
|
||||
// target is a parsed FTP URL. It holds the password, so it must never be
|
||||
// formatted into an error or a log line.
|
||||
type target struct {
|
||||
host string // host name or IP, the TLS server name
|
||||
addr string // host:port
|
||||
user string
|
||||
pass string
|
||||
path string // path relative to the login directory, "" for the directory itself
|
||||
tlsMode string
|
||||
}
|
||||
|
||||
// parseTarget reads an ftp, ftps or ftpes URL. The path follows RFC 1738: it
|
||||
// is relative to the login directory, and an encoded leading slash
|
||||
// (ftp://host/%2Fpub) makes it absolute.
|
||||
func parseTarget(rawURL, optTLS string) (*target, error) {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
// url.Error repeats the whole URL, password included.
|
||||
return nil, errors.New("ftp: invalid URL")
|
||||
}
|
||||
t := &target{}
|
||||
switch strings.ToLower(u.Scheme) {
|
||||
case "ftps":
|
||||
t.tlsMode = pftp.TLSImplicit
|
||||
case "ftpes":
|
||||
t.tlsMode = pftp.TLSExplicit
|
||||
case "ftp":
|
||||
switch strings.ToLower(optTLS) {
|
||||
case "", pftp.TLSNone:
|
||||
t.tlsMode = pftp.TLSNone
|
||||
case pftp.TLSExplicit:
|
||||
t.tlsMode = pftp.TLSExplicit
|
||||
case pftp.TLSImplicit:
|
||||
t.tlsMode = pftp.TLSImplicit
|
||||
default:
|
||||
return nil, fmt.Errorf("ftp: unknown tls mode %q", optTLS)
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf("ftp: unsupported scheme %q", u.Scheme)
|
||||
}
|
||||
t.host = u.Hostname()
|
||||
if t.host == "" {
|
||||
return nil, errors.New("ftp: URL has no host")
|
||||
}
|
||||
port := u.Port()
|
||||
if port == "" {
|
||||
port = "21"
|
||||
if t.tlsMode == pftp.TLSImplicit {
|
||||
port = "990"
|
||||
}
|
||||
}
|
||||
t.addr = net.JoinHostPort(t.host, port)
|
||||
if u.User != nil {
|
||||
// Username and Password are already percent-decoded.
|
||||
t.user = u.User.Username()
|
||||
t.pass, _ = u.User.Password()
|
||||
}
|
||||
if t.user == "" {
|
||||
t.user, t.pass = "anonymous", "anonymous@"
|
||||
}
|
||||
t.path = strings.TrimPrefix(u.Path, "/")
|
||||
return t, nil
|
||||
}
|
||||
|
||||
type dialContextFunc func(ctx context.Context, network, addr string) (net.Conn, error)
|
||||
|
||||
// session is one logged-in control connection. It owns the network
|
||||
// connections it opens, so abort can unblock any read or write in progress.
|
||||
type session struct {
|
||||
*ftp.ServerConn
|
||||
|
||||
ctrl net.Conn // control connection as dialed, under any TLS layer
|
||||
dataHost string // where data connections go
|
||||
|
||||
mu sync.Mutex
|
||||
data net.Conn // current data connection as dialed
|
||||
closed bool
|
||||
stop func() bool
|
||||
}
|
||||
|
||||
// deadline bounds the next control-connection exchange. The control
|
||||
// connection is idle during a transfer, so every exchange sets a new one.
|
||||
func (s *session) deadline() {
|
||||
_ = s.ctrl.SetDeadline(time.Now().Add(readTimeout))
|
||||
}
|
||||
|
||||
// clearDeadline lets the control connection stay idle through a transfer.
|
||||
func (s *session) clearDeadline() {
|
||||
_ = s.ctrl.SetDeadline(time.Time{})
|
||||
}
|
||||
|
||||
// abort closes every connection at once. Pending calls fail immediately.
|
||||
func (s *session) abort() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.closed = true
|
||||
_ = s.ctrl.Close()
|
||||
if s.data != nil {
|
||||
_ = s.data.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// quit logs out and waits, briefly, for the server to hang up. Servers that
|
||||
// cap logins count a session until they have seen it end, so the next login
|
||||
// would otherwise be refused.
|
||||
func (s *session) quit() {
|
||||
if s.stop != nil {
|
||||
s.stop()
|
||||
}
|
||||
_ = s.ctrl.SetDeadline(time.Now().Add(quitTimeout))
|
||||
if s.ServerConn != nil {
|
||||
_ = s.ServerConn.Quit()
|
||||
}
|
||||
s.abort()
|
||||
}
|
||||
|
||||
func (s *session) setData(c net.Conn) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.closed {
|
||||
_ = c.Close()
|
||||
return net.ErrClosed
|
||||
}
|
||||
s.data = c
|
||||
return nil
|
||||
}
|
||||
|
||||
// ctrlConn is the control connection handed to the ftp library.
|
||||
type ctrlConn struct {
|
||||
net.Conn
|
||||
}
|
||||
|
||||
// RemoteAddr must be a *net.TCPAddr: the library asserts it. A connection
|
||||
// through a proxy may report something else, and its address is not used
|
||||
// for data connections anyway (see dialData).
|
||||
func (c ctrlConn) RemoteAddr() net.Addr {
|
||||
if a, ok := c.Conn.RemoteAddr().(*net.TCPAddr); ok {
|
||||
return a
|
||||
}
|
||||
return &net.TCPAddr{}
|
||||
}
|
||||
|
||||
// Close is called by the library's Quit right after it sends QUIT. Reading
|
||||
// until the server hangs up (bounded by the deadline quit set) tells us the
|
||||
// server has ended the session.
|
||||
func (c ctrlConn) Close() error {
|
||||
_, _ = io.Copy(io.Discard, c.Conn)
|
||||
return c.Conn.Close()
|
||||
}
|
||||
|
||||
// dial opens a control connection, upgrades it to TLS when asked, and logs in.
|
||||
//
|
||||
// The library is given a dial function for every connection. That is how
|
||||
// data connections go through the same proxy as the control connection, but
|
||||
// it also means the library does no TLS on them: with a dial function it
|
||||
// neither wraps an implicit-TLS control connection nor any data connection.
|
||||
// We do both here. Explicit TLS (AUTH TLS) is still done by the library, and
|
||||
// so are PBSZ and PROT P after login. For the same reason the library's
|
||||
// DialWithTimeout and DialWithContext would have no effect: connectTimeout
|
||||
// and ctx are applied to our own dials instead.
|
||||
func (f *Fetcher) dial(ctx context.Context, t *target) (*session, error) {
|
||||
netDial, proxied, err := f.netDialer(t)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var tlsConf *tls.Config
|
||||
if t.tlsMode != pftp.TLSNone {
|
||||
tlsConf = f.tlsConfig(t.host)
|
||||
}
|
||||
|
||||
dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
|
||||
defer cancel()
|
||||
raw, err := netDial(dialCtx, "tcp", t.addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctrl := raw
|
||||
if t.tlsMode == pftp.TLSImplicit {
|
||||
tc := tls.Client(raw, tlsConf)
|
||||
if err := tc.HandshakeContext(dialCtx); err != nil {
|
||||
_ = raw.Close()
|
||||
return nil, err
|
||||
}
|
||||
ctrl = tc
|
||||
}
|
||||
|
||||
s := &session{ctrl: raw, dataHost: t.host}
|
||||
if a, ok := raw.RemoteAddr().(*net.TCPAddr); ok && !proxied {
|
||||
// Same IP as the control connection, as the library itself does,
|
||||
// so a name with several addresses cannot send us elsewhere.
|
||||
s.dataHost = a.IP.String()
|
||||
}
|
||||
s.stop = context.AfterFunc(ctx, s.abort)
|
||||
s.deadline()
|
||||
|
||||
handedOut := false
|
||||
opts := []ftp.DialOption{
|
||||
ftp.DialWithShutTimeout(readTimeout),
|
||||
ftp.DialWithDialFunc(func(network, addr string) (net.Conn, error) {
|
||||
if !handedOut {
|
||||
// Dial asks for the control connection first, once.
|
||||
handedOut = true
|
||||
return ctrlConn{ctrl}, nil
|
||||
}
|
||||
return s.dialData(ctx, netDial, tlsConf, addr)
|
||||
}),
|
||||
}
|
||||
switch t.tlsMode {
|
||||
case pftp.TLSImplicit:
|
||||
opts = append(opts, ftp.DialWithTLS(tlsConf))
|
||||
case pftp.TLSExplicit:
|
||||
opts = append(opts, ftp.DialWithExplicitTLS(tlsConf))
|
||||
}
|
||||
c, err := ftp.Dial(t.addr, opts...)
|
||||
if err != nil {
|
||||
s.stop()
|
||||
s.abort()
|
||||
return nil, err
|
||||
}
|
||||
s.ServerConn = c
|
||||
s.deadline()
|
||||
if err := c.Login(t.user, t.pass); err != nil {
|
||||
s.stop()
|
||||
s.abort()
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// dialData opens a passive data connection. The library takes the host from
|
||||
// the control connection's peer, which through a proxy is the proxy, so only
|
||||
// the port is used here.
|
||||
func (s *session) dialData(ctx context.Context, netDial dialContextFunc, tlsConf *tls.Config, addr string) (net.Conn, error) {
|
||||
_, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
|
||||
defer cancel()
|
||||
raw, err := netDial(dialCtx, "tcp", net.JoinHostPort(s.dataHost, port))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.setData(raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tlsConf != nil {
|
||||
// The handshake runs on the first read, as in the library, because
|
||||
// some servers only start it once the transfer command is answered.
|
||||
return tls.Client(raw, tlsConf), nil
|
||||
}
|
||||
return raw, nil
|
||||
}
|
||||
|
||||
// tlsConfig returns the TLS settings for one control connection and its data
|
||||
// connections. Certificates are verified unless the request sets
|
||||
// skipVerifyCert, as for HTTP.
|
||||
//
|
||||
// Each call gets its own one-entry session cache, so a data connection
|
||||
// resumes the TLS session of its own control connection. vsftpd
|
||||
// (require_ssl_reuse) only checks that some session was resumed, but ProFTPD's
|
||||
// mod_tls also checks that it is the control connection's; a cache shared by
|
||||
// several workers would hand a data connection another worker's session.
|
||||
// A cache in the manager's base config is kept as given.
|
||||
func (f *Fetcher) tlsConfig(host string) *tls.Config {
|
||||
cfg := &tls.Config{}
|
||||
if f.manager != nil && f.manager.baseTLS != nil {
|
||||
cfg = f.manager.baseTLS.Clone()
|
||||
}
|
||||
cfg.ServerName = host
|
||||
if f.meta.Req != nil && f.meta.Req.SkipVerifyCert {
|
||||
cfg.InsecureSkipVerify = true
|
||||
}
|
||||
if cfg.ClientSessionCache == nil {
|
||||
cfg.ClientSessionCache = tls.NewLRUClientSessionCache(1)
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// netDialer picks how to reach the server, from the same proxy settings the
|
||||
// HTTP protocol reads (the task's proxy, otherwise the global one).
|
||||
func (f *Fetcher) netDialer(t *target) (dialContextFunc, bool, error) {
|
||||
direct := (&net.Dialer{Timeout: connectTimeout}).DialContext
|
||||
pu, err := f.proxyURL(t)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if pu == nil {
|
||||
return direct, false, nil
|
||||
}
|
||||
switch strings.ToLower(pu.Scheme) {
|
||||
case "socks5", "socks5h":
|
||||
d, err := proxy.FromURL(pu, &net.Dialer{Timeout: connectTimeout})
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("ftp: socks5 proxy: %w", err)
|
||||
}
|
||||
if cd, ok := d.(proxy.ContextDialer); ok {
|
||||
return cd.DialContext, true, nil
|
||||
}
|
||||
return func(_ context.Context, network, addr string) (net.Conn, error) {
|
||||
return d.Dial(network, addr)
|
||||
}, true, nil
|
||||
default:
|
||||
// An HTTP proxy is not used. Every passive data connection goes to
|
||||
// a port the server picks, and HTTP proxies usually allow CONNECT to
|
||||
// port 443 only, so the transfer would fail after the login worked.
|
||||
f.warnOnce.Do(func() {
|
||||
if l := f.logger(); l != nil {
|
||||
l.Warn().Msgf("ftp: the %s proxy at %s cannot carry FTP data connections (HTTP proxy), connecting directly; use a SOCKS5 proxy for FTP", strings.ToLower(pu.Scheme), pu.Host)
|
||||
}
|
||||
})
|
||||
return direct, false, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (f *Fetcher) proxyURL(t *target) (*url.URL, error) {
|
||||
if f.ctl == nil || f.ctl.GetProxy == nil {
|
||||
return nil, nil
|
||||
}
|
||||
handler := f.ctl.GetProxy(f.meta.Req.Proxy)
|
||||
if handler == nil {
|
||||
return nil, nil
|
||||
}
|
||||
// The proxy chooser reads only the URL; the credentials stay out.
|
||||
req := &http.Request{URL: &url.URL{Scheme: "ftp", Host: t.addr, Path: "/" + t.path}, Header: http.Header{}}
|
||||
pu, err := handler(req)
|
||||
if err != nil {
|
||||
return nil, errors.New("ftp: cannot read the proxy settings")
|
||||
}
|
||||
return pu, nil
|
||||
}
|
||||
|
||||
// isLoginRefusal tells a server that is full apart from bad credentials.
|
||||
// 421 always means "not now". 530 means a refused login; it is only a
|
||||
// capacity limit when these credentials have worked before. The library
|
||||
// drops the code when USER itself is refused, so a "too many" message counts
|
||||
// as well.
|
||||
func isLoginRefusal(err error, credentialsWorked bool) bool {
|
||||
var te *textproto.Error
|
||||
if errors.As(err, &te) {
|
||||
switch te.Code {
|
||||
case ftp.StatusNotAvailable:
|
||||
return true
|
||||
case ftp.StatusNotLoggedIn:
|
||||
return credentialsWorked || strings.Contains(strings.ToLower(te.Msg), "too many")
|
||||
}
|
||||
return false
|
||||
}
|
||||
return strings.Contains(strings.ToLower(err.Error()), "too many")
|
||||
}
|
||||
|
||||
// isPermanent reports errors a retry cannot fix: a 5xx reply and a
|
||||
// certificate that does not verify.
|
||||
func isPermanent(err error) bool {
|
||||
var te *textproto.Error
|
||||
if errors.As(err, &te) {
|
||||
return te.Code >= 500
|
||||
}
|
||||
var verifyErr *tls.CertificateVerificationError
|
||||
if errors.As(err, &verifyErr) {
|
||||
return true
|
||||
}
|
||||
var recordErr tls.RecordHeaderError
|
||||
if errors.As(err, &recordErr) {
|
||||
return true
|
||||
}
|
||||
return errors.Is(err, errSizeChanged)
|
||||
}
|
||||
|
||||
func replyCode(err error) int {
|
||||
var te *textproto.Error
|
||||
if errors.As(err, &te) {
|
||||
return te.Code
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -1,146 +0,0 @@
|
||||
package ftp_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rs/zerolog"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/logger"
|
||||
iftp "github.com/GopeedLab/gopeed/internal/protocol/ftp"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
"github.com/GopeedLab/gopeed/pkg/download"
|
||||
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
|
||||
)
|
||||
|
||||
// syncBuffer is a log sink that the downloader's goroutines can share.
|
||||
type syncBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *syncBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *syncBuffer) String() string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.String()
|
||||
}
|
||||
|
||||
// taskEvents records the first done or error event of every task, so the
|
||||
// test never reads a task's status while the downloader writes it.
|
||||
type taskEvents struct {
|
||||
mu sync.Mutex
|
||||
final map[string]download.EventKey
|
||||
}
|
||||
|
||||
func (e *taskEvents) listen(ev *download.Event) {
|
||||
if ev.Key != download.EventKeyDone && ev.Key != download.EventKeyError {
|
||||
return
|
||||
}
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if _, seen := e.final[ev.Task.ID]; !seen {
|
||||
e.final[ev.Task.ID] = ev.Key
|
||||
}
|
||||
}
|
||||
|
||||
func (e *taskEvents) wait(t *testing.T, id string) download.EventKey {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(60 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
e.mu.Lock()
|
||||
key, ok := e.final[id]
|
||||
e.mu.Unlock()
|
||||
if ok {
|
||||
return key
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("task %s did not finish", id)
|
||||
return ""
|
||||
}
|
||||
|
||||
// TestPasswordNeverLogged runs FTP tasks through the downloader with its core
|
||||
// logger captured: a success, a missing file and a wrong password, all behind
|
||||
// an HTTP proxy that FTP must skip with a warning.
|
||||
func TestPasswordNeverLogged(t *testing.T) {
|
||||
srv := iftp.StartFTPServer(t, iftp.ServerOptions{User: "u", Pass: "p@ss"})
|
||||
want := srv.WriteFile(t, "secret.bin", iftp.RandomBytes(300000, 21))
|
||||
|
||||
d := download.NewDownloader(&download.DownloaderConfig{Storage: download.NewMemStorage(), StorageDir: t.TempDir()})
|
||||
logs := &syncBuffer{}
|
||||
d.Logger = &logger.Logger{Logger: zerolog.New(logs)}
|
||||
events := &taskEvents{final: map[string]download.EventKey{}}
|
||||
d.Listener(events.listen)
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Close, not Clear: Clear resets the extension list without a lock while
|
||||
// the last task's done hooks may still read it.
|
||||
defer d.Close()
|
||||
cfg, err := d.GetConfig()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.Proxy = &base.DownloaderProxyConfig{Enable: true, Scheme: "http", Host: "127.0.0.1:1", Usr: "proxyuser", Pwd: "proxysecret"}
|
||||
if err := d.PutConfig(cfg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
dir := t.TempDir()
|
||||
create := func(rawURL string) string {
|
||||
id, err := d.CreateDirect(&base.Request{URL: rawURL}, &base.Options{Path: dir, Extra: &pftp.OptsExtra{Connections: 2}})
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "p@ss") || strings.Contains(err.Error(), "p%40ss") {
|
||||
t.Fatalf("the password leaked into an error: %v", err)
|
||||
}
|
||||
t.Fatal(err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
root := "ftp://u:p%40ss@" + srv.Addr + "/"
|
||||
|
||||
okID := create(root + "secret.bin")
|
||||
if key := events.wait(t, okID); key != download.EventKeyDone {
|
||||
t.Fatalf("download ended with %q, logs:\n%s", key, logs.String())
|
||||
}
|
||||
ok := d.GetTask(okID)
|
||||
if ok.Protocol != "ftp" {
|
||||
t.Fatalf("protocol %q, want ftp", ok.Protocol)
|
||||
}
|
||||
if got := iftp.FileSHA256(t, ok.Meta.SingleFilepath()); got != want {
|
||||
t.Fatalf("sha256 %s, want %s", got, want)
|
||||
}
|
||||
missingID := create(root + "missing.bin")
|
||||
if key := events.wait(t, missingID); key != download.EventKeyError {
|
||||
t.Fatalf("a missing file must fail, got %q", key)
|
||||
}
|
||||
wrongID := create("ftp://u:wr0ng%40pw@" + srv.Addr + "/secret.bin")
|
||||
if key := events.wait(t, wrongID); key != download.EventKeyError {
|
||||
t.Fatalf("a wrong password must fail, got %q", key)
|
||||
}
|
||||
missing, wrong := d.GetTask(missingID), d.GetTask(wrongID)
|
||||
for _, task := range []*download.Task{ok, missing, wrong} {
|
||||
if name := task.Name(); strings.Contains(name, "p@ss") || strings.Contains(name, "p%40ss") || strings.Contains(name, "wr0ng") {
|
||||
t.Fatalf("task name %q shows the password", name)
|
||||
}
|
||||
}
|
||||
|
||||
out := logs.String()
|
||||
if !strings.Contains(out, "task id") || !strings.Contains(out, "HTTP proxy") {
|
||||
t.Fatalf("the captured log misses the expected lines, so it proves nothing:\n%s", out)
|
||||
}
|
||||
for _, secret := range []string{"p@ss", "p%40ss", "wr0ng", "proxysecret"} {
|
||||
if strings.Contains(out, secret) {
|
||||
t.Fatalf("%q appears in the log:\n%s", secret, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -1,109 +0,0 @@
|
||||
package ftp
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"net/url"
|
||||
"path"
|
||||
"strings"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
|
||||
)
|
||||
|
||||
// fetcherData is what a task saves to resume later. Segments has the same
|
||||
// Begin, End and Downloaded fields every version writes; the other fields are
|
||||
// optional, so a save without them still restores.
|
||||
type fetcherData struct {
|
||||
Segments []*segment
|
||||
// MaxConns is the login ceiling the server enforced, 0 when none was hit.
|
||||
MaxConns int `json:",omitempty"`
|
||||
// NoRest records that the server refused REST, so the task downloads each
|
||||
// file in one sequential stream.
|
||||
NoRest bool `json:",omitempty"`
|
||||
}
|
||||
|
||||
// FetcherManager handles ftp://, ftps:// (implicit TLS) and ftpes:// (explicit
|
||||
// TLS) URLs.
|
||||
type FetcherManager struct {
|
||||
// baseTLS is the base for every TLS connection. Nil means the system
|
||||
// roots; the package's tests set it to trust their own certificate. The
|
||||
// task's skipVerifyCert still turns verification off.
|
||||
baseTLS *tls.Config
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Name() string {
|
||||
return "ftp"
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Filters() []*fetcher.SchemeFilter {
|
||||
return []*fetcher.SchemeFilter{
|
||||
{Type: fetcher.FilterTypeUrl, Pattern: "FTP"},
|
||||
{Type: fetcher.FilterTypeUrl, Pattern: "FTPS"},
|
||||
{Type: fetcher.FilterTypeUrl, Pattern: "FTPES"},
|
||||
}
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Build() fetcher.Fetcher {
|
||||
return &Fetcher{manager: fm}
|
||||
}
|
||||
|
||||
// ParseName names a task before it is resolved. It never includes the
|
||||
// userinfo of the URL.
|
||||
func (fm *FetcherManager) ParseName(u string) string {
|
||||
parsed, err := url.Parse(u)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
name := path.Base(strings.TrimSuffix(parsed.Path, "/"))
|
||||
if name == "" || name == "/" || name == "." {
|
||||
name = parsed.Hostname()
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) AutoRename() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) DefaultConfig() any {
|
||||
return &config{Connections: defaultConnections}
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Store(f fetcher.Fetcher) (any, error) {
|
||||
ff := f.(*Fetcher)
|
||||
ff.mu.Lock()
|
||||
defer ff.mu.Unlock()
|
||||
segments := make([]*segment, len(ff.segments))
|
||||
for i, s := range ff.segments {
|
||||
c := *s
|
||||
segments[i] = &c
|
||||
}
|
||||
return &fetcherData{
|
||||
Segments: segments,
|
||||
MaxConns: ff.maxConns,
|
||||
NoRest: ff.noRest,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) {
|
||||
return &fetcherData{}, func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher {
|
||||
fd := v.(*fetcherData)
|
||||
ff := fm.Build().(*Fetcher)
|
||||
ff.meta = meta
|
||||
if meta != nil && meta.Req != nil {
|
||||
_ = base.ParseReqExtra[pftp.ReqExtra](meta.Req)
|
||||
}
|
||||
if meta != nil && meta.Opts != nil {
|
||||
_ = base.ParseOptExtra[pftp.OptsExtra](meta.Opts)
|
||||
}
|
||||
ff.segments = fd.Segments
|
||||
ff.maxConns = fd.MaxConns
|
||||
ff.noRest = fd.NoRest
|
||||
return ff
|
||||
}
|
||||
}
|
||||
|
||||
func (fm *FetcherManager) Close() error {
|
||||
return nil
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
File diff suppressed because it is too large.
Load diff
@@ -1,187 +0,0 @@
|
||||
package ftp
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"fmt"
|
||||
"path"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jlaffaye/ftp"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
)
|
||||
|
||||
// Limits of a directory download. A tree past either one is refused with an
|
||||
// error that names the limit; it is never cut short.
|
||||
const maxWalkDepth = 8
|
||||
|
||||
// maxWalkFiles is a variable so tests can lower it.
|
||||
var maxWalkFiles = 10000
|
||||
|
||||
// resolveTarget describes the URL's path: one file, or a directory walked
|
||||
// into a multi-file resource whose paths are relative to it.
|
||||
func (f *Fetcher) resolveTarget(s *session, t *target) (*base.Resource, error) {
|
||||
isDir, size, mtime, err := f.stat(s, t.path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !isDir {
|
||||
return &base.Resource{
|
||||
Size: size,
|
||||
Range: true,
|
||||
Files: []*base.FileInfo{{
|
||||
Name: path.Base(t.path),
|
||||
Size: size,
|
||||
Ctime: mtime,
|
||||
}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
files, err := f.walk(s, t.path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
name := path.Base(strings.TrimSuffix(t.path, "/"))
|
||||
if name == "" || name == "." || name == "/" {
|
||||
name = t.host
|
||||
}
|
||||
res := &base.Resource{Name: name, Range: true, Files: files}
|
||||
res.CalcSize(nil)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// stat tells a file from a directory. MLST answers both at once. Without
|
||||
// it, SIZE answers for a file, and a directory is one we can change into.
|
||||
func (f *Fetcher) stat(s *session, p string) (isDir bool, size int64, mtime *time.Time, err error) {
|
||||
if p == "" || strings.HasSuffix(p, "/") {
|
||||
return true, 0, nil, nil
|
||||
}
|
||||
s.deadline()
|
||||
if entry, mlstErr := s.GetEntry(p); mlstErr == nil {
|
||||
switch entry.Type {
|
||||
case ftp.EntryTypeFolder:
|
||||
return true, 0, nil, nil
|
||||
case ftp.EntryTypeFile:
|
||||
if !entry.Time.IsZero() {
|
||||
t := entry.Time
|
||||
mtime = &t
|
||||
}
|
||||
return false, int64(entry.Size), mtime, nil
|
||||
}
|
||||
// A link: SIZE below follows it on the server.
|
||||
}
|
||||
// MLST is missing or refused the path: the older commands decide. A
|
||||
// broken connection fails them too, with its own error.
|
||||
|
||||
s.deadline()
|
||||
size, sizeErr := s.FileSize(p)
|
||||
if sizeErr == nil {
|
||||
if s.IsGetTimeSupported() {
|
||||
s.deadline()
|
||||
if t, err := s.GetTime(p); err == nil {
|
||||
mtime = &t
|
||||
}
|
||||
}
|
||||
return false, size, mtime, nil
|
||||
}
|
||||
if replyCode(sizeErr) == 0 {
|
||||
return false, 0, nil, sizeErr
|
||||
}
|
||||
|
||||
// SIZE failed: a directory, or a missing path.
|
||||
s.deadline()
|
||||
cwd, err := s.CurrentDir()
|
||||
if err != nil {
|
||||
return false, 0, nil, err
|
||||
}
|
||||
s.deadline()
|
||||
if err := s.ChangeDir(p); err != nil {
|
||||
return false, 0, nil, sizeErr
|
||||
}
|
||||
s.deadline()
|
||||
if err := s.ChangeDir(cwd); err != nil {
|
||||
return false, 0, nil, err
|
||||
}
|
||||
return true, 0, nil, nil
|
||||
}
|
||||
|
||||
// safeEntryName reports whether a name from a listing can become a local
|
||||
// path element. A hostile server could otherwise list "../x" and write
|
||||
// outside the task directory.
|
||||
func safeEntryName(name string) bool {
|
||||
if name == "" || name == "." || name == ".." {
|
||||
return false
|
||||
}
|
||||
return !strings.ContainsAny(name, "/\\\x00")
|
||||
}
|
||||
|
||||
// walk lists a directory tree, at most maxWalkDepth levels below root and
|
||||
// maxWalkFiles files. Files come back sorted by path, then name.
|
||||
func (f *Fetcher) walk(s *session, root string) ([]*base.FileInfo, error) {
|
||||
type dir struct {
|
||||
remote string // path on the server
|
||||
rel string // path below root, "" for root itself
|
||||
depth int
|
||||
}
|
||||
var files []*base.FileInfo
|
||||
stack := []dir{{remote: strings.TrimSuffix(root, "/"), rel: "", depth: 0}}
|
||||
for len(stack) > 0 {
|
||||
d := stack[len(stack)-1]
|
||||
stack = stack[:len(stack)-1]
|
||||
|
||||
s.deadline()
|
||||
entries, err := s.List(d.remote)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.Name == "." || e.Name == ".." {
|
||||
continue
|
||||
}
|
||||
if !safeEntryName(e.Name) {
|
||||
f.warn("ftp: skipping a listed name that is not a safe file name: %q", e.Name)
|
||||
continue
|
||||
}
|
||||
remote := path.Join(d.remote, e.Name)
|
||||
rel := path.Join(d.rel, e.Name)
|
||||
if d.remote == "" {
|
||||
remote = e.Name
|
||||
}
|
||||
entryType, size := e.Type, int64(e.Size)
|
||||
if entryType == ftp.EntryTypeLink {
|
||||
// A link to a file has a size; a link to a directory is
|
||||
// not followed, so a loop cannot make the walk endless.
|
||||
s.deadline()
|
||||
linkSize, err := s.FileSize(remote)
|
||||
if err != nil {
|
||||
f.warn("ftp: skipping link %q: %v", rel, err)
|
||||
continue
|
||||
}
|
||||
entryType, size = ftp.EntryTypeFile, linkSize
|
||||
}
|
||||
switch entryType {
|
||||
case ftp.EntryTypeFolder:
|
||||
if d.depth+1 > maxWalkDepth {
|
||||
return nil, fmt.Errorf("ftp: the folder is deeper than %d levels, the limit for a folder download", maxWalkDepth)
|
||||
}
|
||||
stack = append(stack, dir{remote: remote, rel: rel, depth: d.depth + 1})
|
||||
case ftp.EntryTypeFile:
|
||||
if len(files) >= maxWalkFiles {
|
||||
return nil, fmt.Errorf("ftp: the folder has more than %d files, the limit for a folder download", maxWalkFiles)
|
||||
}
|
||||
fi := &base.FileInfo{Name: e.Name, Path: d.rel, Size: size}
|
||||
if !e.Time.IsZero() {
|
||||
t := e.Time
|
||||
fi.Ctime = &t
|
||||
}
|
||||
files = append(files, fi)
|
||||
}
|
||||
}
|
||||
}
|
||||
slices.SortFunc(files, func(a, b *base.FileInfo) int {
|
||||
return cmp.Or(strings.Compare(a.Path, b.Path), strings.Compare(a.Name, b.Name))
|
||||
})
|
||||
return files, nil
|
||||
}
|
||||
@@ -303,9 +303,8 @@ type Fetcher struct {
|
||||
resolveDataPos atomic.Int64 // How many bytes downloaded during resolve
|
||||
|
||||
// Resolve response - kept open for one-time URLs
|
||||
resolveResp *http.Response
|
||||
resolveRespCancel context.CancelFunc // Cancels the Resolve request
|
||||
resolveRespLock sync.Mutex
|
||||
resolveResp *http.Response
|
||||
resolveRespLock sync.Mutex
|
||||
|
||||
// Async prefetch during resolve phase
|
||||
prefetchFile *os.File // Temporary file for prefetch data
|
||||
@@ -426,11 +425,8 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
|
||||
// Build HTTP request WITHOUT Range header (normal request)
|
||||
// This allows the response to be reused for downloading (important for one-time URLs)
|
||||
// The body outlives Resolve, so the read timeout gets its own cancel func.
|
||||
reqCtx, reqCancel := context.WithCancel(context.Background())
|
||||
httpReq, err := f.buildRequest(reqCtx, req)
|
||||
httpReq, err := f.buildRequest(context.TODO(), req)
|
||||
if err != nil {
|
||||
reqCancel()
|
||||
f.setState(stateError)
|
||||
return err
|
||||
}
|
||||
@@ -442,7 +438,6 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
connStartTime := time.Now()
|
||||
resp, err := client.Do(httpReq)
|
||||
if err != nil {
|
||||
reqCancel()
|
||||
f.setState(stateError)
|
||||
return err
|
||||
}
|
||||
@@ -457,7 +452,6 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
|
||||
if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent {
|
||||
resp.Body.Close()
|
||||
reqCancel()
|
||||
f.setState(stateError)
|
||||
return NewRequestError(resp.StatusCode)
|
||||
}
|
||||
@@ -475,7 +469,6 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
start, _, total, err := parseRangeResponse(resp)
|
||||
if err != nil {
|
||||
resp.Body.Close()
|
||||
reqCancel()
|
||||
f.setState(stateError)
|
||||
return err
|
||||
}
|
||||
@@ -543,10 +536,8 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
f.resolveRespLock.Lock()
|
||||
if reuseResolveBody {
|
||||
f.resolveResp = resp
|
||||
f.resolveRespCancel = reqCancel
|
||||
} else {
|
||||
resp.Body.Close()
|
||||
reqCancel()
|
||||
f.prefetchDone.Store(true)
|
||||
}
|
||||
f.resolveRespLock.Unlock()
|
||||
@@ -583,7 +574,7 @@ func (f *Fetcher) asyncPrefetch() {
|
||||
|
||||
// Get the resolve response
|
||||
f.resolveRespLock.Lock()
|
||||
resp, cancel := f.resolveResp, f.resolveRespCancel
|
||||
resp := f.resolveResp
|
||||
f.resolveRespLock.Unlock()
|
||||
|
||||
if resp == nil {
|
||||
@@ -594,7 +585,10 @@ func (f *Fetcher) asyncPrefetch() {
|
||||
// Close the response even if temporary-file setup fails or shutdown
|
||||
// rejects a late prefetch. Otherwise the server can remain blocked writing.
|
||||
f.resolveRespLock.Lock()
|
||||
f.closeResolveRespLocked()
|
||||
if f.resolveResp != nil {
|
||||
f.resolveResp.Body.Close()
|
||||
f.resolveResp = nil
|
||||
}
|
||||
f.resolveRespLock.Unlock()
|
||||
}()
|
||||
|
||||
@@ -624,7 +618,7 @@ func (f *Fetcher) asyncPrefetch() {
|
||||
f.prefetchFilePath = tmpFile.Name()
|
||||
|
||||
buf := make([]byte, 32*1024) // 32KB buffer
|
||||
reader := f.responseReader(resp, cancel)
|
||||
reader := f.responseReader(resp)
|
||||
|
||||
for {
|
||||
select {
|
||||
@@ -887,7 +881,10 @@ func (f *Fetcher) doStart() error {
|
||||
if !f.resolveFallback.Load() {
|
||||
// Also close resolve response if still open.
|
||||
f.resolveRespLock.Lock()
|
||||
f.closeResolveRespLocked()
|
||||
if f.resolveResp != nil {
|
||||
f.resolveResp.Body.Close()
|
||||
f.resolveResp = nil
|
||||
}
|
||||
f.resolveRespLock.Unlock()
|
||||
}
|
||||
}
|
||||
@@ -1340,11 +1337,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
|
||||
}
|
||||
f.connMu.Unlock()
|
||||
|
||||
// The read timeout cancels this request through its own context.
|
||||
reqCtx, reqCancel := context.WithCancel(conn.ctx)
|
||||
defer reqCancel()
|
||||
|
||||
httpReq, err := f.buildRequest(reqCtx, f.meta.Req)
|
||||
httpReq, err := f.buildRequest(conn.ctx, f.meta.Req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -1372,7 +1365,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
|
||||
// Check if this might be a redirect URL expiration error
|
||||
// If so, try falling back to the original URL
|
||||
if f.hasRedirectURL() && isRedirectExpiredError(originalErr) {
|
||||
fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(reqCtx, client, rangeStart, rangeEnd, rangeRequested, ifRange)
|
||||
fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(conn.ctx, client, rangeStart, rangeEnd, rangeRequested, ifRange)
|
||||
if fallbackErr == nil && fallbackResp != nil {
|
||||
// Fallback succeeded, use this response instead
|
||||
resp = fallbackResp
|
||||
@@ -1493,7 +1486,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
|
||||
f.slowStart.onConnectSuccess()
|
||||
}
|
||||
|
||||
reader := f.responseReader(resp, reqCancel)
|
||||
reader := f.responseReader(resp)
|
||||
var responseBytesRead int64
|
||||
for {
|
||||
if conn.ctx.Err() != nil {
|
||||
@@ -1847,8 +1840,8 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
|
||||
|
||||
// Get the resolve response
|
||||
f.resolveRespLock.Lock()
|
||||
resp, cancel := f.resolveResp, f.resolveRespCancel
|
||||
f.resolveResp, f.resolveRespCancel = nil, nil // Take ownership
|
||||
resp := f.resolveResp
|
||||
f.resolveResp = nil // Take ownership
|
||||
f.resolveRespLock.Unlock()
|
||||
|
||||
if resp == nil {
|
||||
@@ -1857,9 +1850,6 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
|
||||
return
|
||||
}
|
||||
|
||||
if cancel != nil {
|
||||
defer cancel()
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
f.connMu.Lock()
|
||||
@@ -1876,7 +1866,7 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
|
||||
}
|
||||
|
||||
// Download data from resolve response
|
||||
reader := f.responseReader(resp, cancel)
|
||||
reader := f.responseReader(resp)
|
||||
for {
|
||||
if conn.ctx.Err() != nil {
|
||||
return
|
||||
@@ -2491,7 +2481,10 @@ func (f *Fetcher) Pause() error {
|
||||
|
||||
// Clean up resolve response if still held
|
||||
f.resolveRespLock.Lock()
|
||||
f.closeResolveRespLocked()
|
||||
if f.resolveResp != nil {
|
||||
f.resolveResp.Body.Close()
|
||||
f.resolveResp = nil
|
||||
}
|
||||
f.resolveRespLock.Unlock()
|
||||
|
||||
f.fileMu.Lock()
|
||||
@@ -2609,25 +2602,9 @@ func (f *Fetcher) Wait() error {
|
||||
|
||||
// Generated output can pause while its producer waits for processing capacity.
|
||||
// Its producer owns upstream deadlines; cancellation still closes the response.
|
||||
// Otherwise a stalled read cancels the request via cancel, which must belong
|
||||
// to the request that produced resp.
|
||||
func (f *Fetcher) responseReader(resp *http.Response, cancel context.CancelFunc) io.Reader {
|
||||
func (f *Fetcher) responseReader(resp *http.Response) io.Reader {
|
||||
if f.ctl != nil && f.ctl.ManagedProduction != nil && resp.Request != nil && f.ctl.ManagedProduction(resp.Request.URL.String()) {
|
||||
return resp.Body
|
||||
}
|
||||
return NewTimeoutReader(resp.Body, readTimeout, cancel)
|
||||
}
|
||||
|
||||
// closeResolveRespLocked closes the kept Resolve response, if any, and
|
||||
// releases its request context. The caller must hold resolveRespLock.
|
||||
func (f *Fetcher) closeResolveRespLocked() {
|
||||
if f.resolveResp == nil {
|
||||
return
|
||||
}
|
||||
f.resolveResp.Body.Close()
|
||||
f.resolveResp = nil
|
||||
if f.resolveRespCancel != nil {
|
||||
f.resolveRespCancel()
|
||||
f.resolveRespCancel = nil
|
||||
}
|
||||
return NewTimeoutReader(resp.Body, readTimeout)
|
||||
}
|
||||
@@ -6,56 +6,35 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// TimeoutReader fails a Read that makes no progress within timeout.
|
||||
//
|
||||
// One timer per reader is armed when Read starts and stopped before it
|
||||
// returns, so time the caller spends between reads is not counted. If the
|
||||
// timer fires, it calls cancel, which should cancel the HTTP request the
|
||||
// reader belongs to so the transport unblocks the pending Read. Read still
|
||||
// waits for the underlying Read to return; that Read and every later one
|
||||
// return context.DeadlineExceeded. Errors from any other cancellation are
|
||||
// returned unchanged, and closing the response body is left to the caller.
|
||||
//
|
||||
// With a nil cancel the pending Read cannot be interrupted; it still reports
|
||||
// context.DeadlineExceeded, but only once it returns.
|
||||
//
|
||||
// TimeoutReader is not safe for concurrent use.
|
||||
type TimeoutReader struct {
|
||||
reader io.Reader
|
||||
timeout time.Duration
|
||||
cancel context.CancelFunc
|
||||
timer *time.Timer
|
||||
timedOut bool
|
||||
reader io.Reader
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
func NewTimeoutReader(r io.Reader, timeout time.Duration, cancel context.CancelFunc) *TimeoutReader {
|
||||
if cancel == nil {
|
||||
cancel = func() {}
|
||||
}
|
||||
func NewTimeoutReader(r io.Reader, timeout time.Duration) *TimeoutReader {
|
||||
return &TimeoutReader{
|
||||
reader: r,
|
||||
timeout: timeout,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
func (tr *TimeoutReader) Read(p []byte) (n int, err error) {
|
||||
if tr.timedOut {
|
||||
return 0, context.DeadlineExceeded
|
||||
}
|
||||
if tr.timer == nil {
|
||||
tr.timer = time.AfterFunc(tr.timeout, tr.cancel)
|
||||
} else {
|
||||
tr.timer.Reset(tr.timeout)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), tr.timeout)
|
||||
defer cancel()
|
||||
|
||||
n, err = tr.reader.Read(p)
|
||||
done := make(chan struct{})
|
||||
var readErr error
|
||||
var bytesRead int
|
||||
|
||||
if !tr.timer.Stop() {
|
||||
// The timer fired during this Read and cancelled the request, so err is
|
||||
// whatever the interrupted read produced. Report the timeout instead.
|
||||
tr.timedOut = true
|
||||
return n, context.DeadlineExceeded
|
||||
go func() {
|
||||
bytesRead, readErr = tr.reader.Read(p)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return bytesRead, readErr
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
@@ -2,26 +2,17 @@ package http
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"errors"
|
||||
"io"
|
||||
gohttp "net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/httpclient"
|
||||
)
|
||||
|
||||
func TestTimeoutReader_Read(t *testing.T) {
|
||||
data := []byte("Hello, World!")
|
||||
reader := bytes.NewReader(data)
|
||||
timeoutReader := NewTimeoutReader(reader, 1*time.Second, nil)
|
||||
timeoutReader := NewTimeoutReader(reader, 1*time.Second)
|
||||
|
||||
buf := make([]byte, len(data))
|
||||
n, err := timeoutReader.Read(buf)
|
||||
@@ -38,7 +29,7 @@ func TestTimeoutReader_Read(t *testing.T) {
|
||||
|
||||
func TestTimeoutReader_ReadTimeout(t *testing.T) {
|
||||
reader := &slowReader{delay: 2 * time.Second}
|
||||
timeoutReader := NewTimeoutReader(reader, 1*time.Second, nil)
|
||||
timeoutReader := NewTimeoutReader(reader, 1*time.Second)
|
||||
|
||||
buf := make([]byte, 8192)
|
||||
_, err := timeoutReader.Read(buf)
|
||||
@@ -58,340 +49,3 @@ func (sr *slowReader) Read(p []byte) (n int, err error) {
|
||||
time.Sleep(sr.delay)
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
// The timer must only cover time spent inside Read, not caller work between reads.
|
||||
func TestTimeoutReader_IdleBetweenReads(t *testing.T) {
|
||||
var cancelled atomic.Bool
|
||||
timeoutReader := NewTimeoutReader(bytes.NewReader([]byte("abcdef")), 50*time.Millisecond, func() {
|
||||
cancelled.Store(true)
|
||||
})
|
||||
|
||||
buf := make([]byte, 3)
|
||||
if _, err := timeoutReader.Read(buf); err != nil {
|
||||
t.Fatalf("first read: %v", err)
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
if _, err := timeoutReader.Read(buf); err != nil {
|
||||
t.Fatalf("second read after idle: %v", err)
|
||||
}
|
||||
if cancelled.Load() {
|
||||
t.Fatal("request was cancelled while idle between reads")
|
||||
}
|
||||
}
|
||||
|
||||
// A timed-out Read must not return while the underlying Read can still write
|
||||
// into the caller's buffer, and must leave closing the reader to the caller.
|
||||
func TestTimeoutReader_TimedOutReadDoesNotOutliveCaller(t *testing.T) {
|
||||
reader := &lateWriteReader{cancelled: make(chan struct{})}
|
||||
timeoutReader := NewTimeoutReader(reader, 50*time.Millisecond, reader.cancel)
|
||||
|
||||
buf := make([]byte, 8)
|
||||
_, err := timeoutReader.Read(buf)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
|
||||
}
|
||||
if !reader.returned.Load() {
|
||||
t.Fatal("Read returned before the underlying Read finished")
|
||||
}
|
||||
if _, err := timeoutReader.Read(buf); !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("read after timeout: expected %v, got %v", context.DeadlineExceeded, err)
|
||||
}
|
||||
if reader.closed.Load() {
|
||||
t.Fatal("reader was closed by the timeout")
|
||||
}
|
||||
}
|
||||
|
||||
// lateWriteReader blocks until cancelled, then writes into p before returning.
|
||||
type lateWriteReader struct {
|
||||
cancelOnce sync.Once
|
||||
cancelled chan struct{}
|
||||
returned atomic.Bool
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
func (r *lateWriteReader) Read(p []byte) (int, error) {
|
||||
<-r.cancelled
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
p[0] = 'x'
|
||||
r.returned.Store(true)
|
||||
return 0, context.Canceled
|
||||
}
|
||||
|
||||
func (r *lateWriteReader) cancel() {
|
||||
r.cancelOnce.Do(func() { close(r.cancelled) })
|
||||
}
|
||||
|
||||
func (r *lateWriteReader) Close() error {
|
||||
r.closed.Store(true)
|
||||
return nil
|
||||
}
|
||||
|
||||
type timeoutReaderClientCase struct {
|
||||
name string
|
||||
http2 bool
|
||||
client func(t *testing.T, server *httptest.Server) *gohttp.Client
|
||||
}
|
||||
|
||||
func timeoutReaderClientCases() []timeoutReaderClientCase {
|
||||
newClient := func(mode httpclient.ImpersonationMode) func(t *testing.T, server *httptest.Server) *gohttp.Client {
|
||||
return func(t *testing.T, server *httptest.Server) *gohttp.Client {
|
||||
var tlsConfig *tls.Config
|
||||
if server.TLS != nil {
|
||||
roots := x509.NewCertPool()
|
||||
roots.AddCert(server.Certificate())
|
||||
tlsConfig = &tls.Config{RootCAs: roots}
|
||||
}
|
||||
client, err := httpclient.NewClient(httpclient.Options{
|
||||
Transport: httpclient.TransportOptions{
|
||||
TLSClientConfig: tlsConfig,
|
||||
},
|
||||
Impersonation: httpclient.ImpersonationOptions{
|
||||
Mode: mode,
|
||||
Browser: httpclient.BrowserChrome,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return client
|
||||
}
|
||||
}
|
||||
var cases []timeoutReaderClientCase
|
||||
for _, mode := range []httpclient.ImpersonationMode{httpclient.ImpersonationNative, httpclient.ImpersonationFixed} {
|
||||
cases = append(cases,
|
||||
timeoutReaderClientCase{name: string(mode) + "/http1", client: newClient(mode)},
|
||||
timeoutReaderClientCase{name: string(mode) + "/http2", http2: true, client: newClient(mode)},
|
||||
)
|
||||
}
|
||||
return cases
|
||||
}
|
||||
|
||||
func startTimeoutReaderServer(t *testing.T, useHTTP2 bool, handler gohttp.HandlerFunc) *httptest.Server {
|
||||
server := httptest.NewUnstartedServer(handler)
|
||||
if useHTTP2 {
|
||||
server.EnableHTTP2 = true
|
||||
server.StartTLS()
|
||||
} else {
|
||||
server.Start()
|
||||
}
|
||||
t.Cleanup(server.Close)
|
||||
return server
|
||||
}
|
||||
|
||||
// readBodyWithTimeout sends the request under its own context derived from
|
||||
// ctx, as the fetcher does, and reads the body through a TimeoutReader.
|
||||
func readBodyWithTimeout(t *testing.T, ctx context.Context, client *gohttp.Client, url string, timeout time.Duration, wantProto int) ([]byte, error) {
|
||||
t.Helper()
|
||||
reqCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
req, err := gohttp.NewRequestWithContext(reqCtx, gohttp.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.ProtoMajor != wantProto {
|
||||
t.Fatalf("expected HTTP/%d, got %s", wantProto, resp.Proto)
|
||||
}
|
||||
|
||||
type result struct {
|
||||
data []byte
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
data, err := io.ReadAll(NewTimeoutReader(resp.Body, timeout, cancel))
|
||||
done <- result{data, err}
|
||||
}()
|
||||
select {
|
||||
case r := <-done:
|
||||
return r.data, r.err
|
||||
case <-time.After(timeout + 10*time.Second):
|
||||
t.Fatal("Read did not return after the read timeout")
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
func TestTimeoutReader_StalledHTTPBodyTimesOut(t *testing.T) {
|
||||
for _, tc := range timeoutReaderClientCases() {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
|
||||
w.Header().Set("Content-Length", "1048576")
|
||||
w.Write(bytes.Repeat([]byte("a"), 1024))
|
||||
w.(gohttp.Flusher).Flush()
|
||||
<-r.Context().Done()
|
||||
})
|
||||
|
||||
wantProto := 1
|
||||
if tc.http2 {
|
||||
wantProto = 2
|
||||
}
|
||||
const timeout = 200 * time.Millisecond
|
||||
start := time.Now()
|
||||
data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, timeout, wantProto)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
|
||||
}
|
||||
if len(data) != 1024 {
|
||||
t.Fatalf("expected the 1024 bytes sent before the stall, got %d", len(data))
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed < timeout {
|
||||
t.Fatalf("timed out after %v, before the %v read timeout", elapsed, timeout)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTimeoutReader_SlowSteadyHTTPBodyDoesNotTimeOut(t *testing.T) {
|
||||
const (
|
||||
timeout = 300 * time.Millisecond
|
||||
interval = 30 * time.Millisecond
|
||||
chunks = 40 // 1.2s in total, four times the read timeout
|
||||
)
|
||||
for _, tc := range timeoutReaderClientCases() {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
|
||||
for i := 0; i < chunks; i++ {
|
||||
w.Write([]byte("0123456789"))
|
||||
w.(gohttp.Flusher).Flush()
|
||||
time.Sleep(interval)
|
||||
}
|
||||
})
|
||||
|
||||
wantProto := 1
|
||||
if tc.http2 {
|
||||
wantProto = 2
|
||||
}
|
||||
data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, timeout, wantProto)
|
||||
if err != nil {
|
||||
t.Fatalf("expected no error, got %v", err)
|
||||
}
|
||||
if len(data) != chunks*10 {
|
||||
t.Fatalf("expected %d bytes, got %d", chunks*10, len(data))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A gzip-encoded body that stalls must time out without racing the caller's
|
||||
// Body.Close against the decompressing reader.
|
||||
func TestTimeoutReader_StalledGzipHTTPBodyTimesOut(t *testing.T) {
|
||||
payload := bytes.Repeat([]byte("gopeed"), 1024)
|
||||
stages := []struct {
|
||||
name string
|
||||
write func(w io.Writer)
|
||||
want int
|
||||
}{
|
||||
{
|
||||
// The first bytes of a gzip header, so the client blocks inside
|
||||
// gzip.NewReader.
|
||||
name: "incomplete-header",
|
||||
write: func(w io.Writer) { w.Write([]byte{0x1f, 0x8b, 0x08, 0x00}) },
|
||||
want: 0,
|
||||
},
|
||||
{
|
||||
// A complete header and some compressed data, flushed without
|
||||
// finishing the gzip stream.
|
||||
name: "stalled-body",
|
||||
write: func(w io.Writer) {
|
||||
gz := gzip.NewWriter(w)
|
||||
gz.Write(payload)
|
||||
gz.Flush()
|
||||
},
|
||||
want: len(payload),
|
||||
},
|
||||
}
|
||||
for _, stage := range stages {
|
||||
for _, tc := range timeoutReaderClientCases() {
|
||||
t.Run(stage.name+"/"+tc.name, func(t *testing.T) {
|
||||
server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
|
||||
w.Header().Set("Content-Encoding", "gzip")
|
||||
stage.write(w)
|
||||
w.(gohttp.Flusher).Flush()
|
||||
<-r.Context().Done()
|
||||
})
|
||||
|
||||
wantProto := 1
|
||||
if tc.http2 {
|
||||
wantProto = 2
|
||||
}
|
||||
data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, 200*time.Millisecond, wantProto)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
|
||||
}
|
||||
if len(data) != stage.want {
|
||||
t.Fatalf("expected the %d bytes sent before the stall, got %d", stage.want, len(data))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Cancelling the caller's context must not be reported as a read timeout.
|
||||
func TestTimeoutReader_CallerCancellationIsNotTimeout(t *testing.T) {
|
||||
for _, tc := range timeoutReaderClientCases() {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
|
||||
w.Header().Set("Content-Length", "1048576")
|
||||
w.Write(bytes.Repeat([]byte("a"), 1024))
|
||||
w.(gohttp.Flusher).Flush()
|
||||
<-r.Context().Done()
|
||||
})
|
||||
|
||||
wantProto := 1
|
||||
if tc.http2 {
|
||||
wantProto = 2
|
||||
}
|
||||
// Cancel once the client has read the first bytes, so the next
|
||||
// Read is the one blocked when the caller cancels.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
client := tc.client(t, server)
|
||||
next := client.Transport
|
||||
client.Transport = roundTripFunc(func(r *gohttp.Request) (*gohttp.Response, error) {
|
||||
resp, err := next.RoundTrip(r)
|
||||
if err == nil {
|
||||
resp.Body = &cancelAfterFirstRead{ReadCloser: resp.Body, cancel: cancel}
|
||||
}
|
||||
return resp, err
|
||||
})
|
||||
|
||||
const timeout = 5 * time.Second
|
||||
start := time.Now()
|
||||
data, err := readBodyWithTimeout(t, ctx, client, server.URL, timeout, wantProto)
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("expected %v, got %v", context.Canceled, err)
|
||||
}
|
||||
if len(data) == 0 {
|
||||
t.Fatal("cancelled before any data was read")
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed >= timeout {
|
||||
t.Fatalf("read returned after %v, not on cancellation", elapsed)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type roundTripFunc func(*gohttp.Request) (*gohttp.Response, error)
|
||||
|
||||
func (f roundTripFunc) RoundTrip(r *gohttp.Request) (*gohttp.Response, error) {
|
||||
return f(r)
|
||||
}
|
||||
|
||||
// cancelAfterFirstRead calls cancel once a Read has returned data.
|
||||
type cancelAfterFirstRead struct {
|
||||
io.ReadCloser
|
||||
once sync.Once
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
func (b *cancelAfterFirstRead) Read(p []byte) (int, error) {
|
||||
n, err := b.ReadCloser.Read(p)
|
||||
if n > 0 {
|
||||
b.once.Do(b.cancel)
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
+2
-2
@@ -180,11 +180,11 @@ func (s *Service) createTaskBatch(ctx *Context) *Response {
|
||||
if len(request.Reqs) == 0 {
|
||||
return newErrorResponse("param invalid: reqs", restmodel.CodeInvalidParam)
|
||||
}
|
||||
result, err := s.Downloader.CreateDirectBatchWithResult(&request)
|
||||
taskIDs, err := s.Downloader.CreateDirectBatch(&request)
|
||||
if err != nil {
|
||||
return newErrorResponse(err.Error())
|
||||
}
|
||||
return newOKResponse(result)
|
||||
return newOKResponse(taskIDs)
|
||||
}
|
||||
|
||||
func (s *Service) patchTask(ctx *Context) *Response {
|
||||
|
||||
@@ -267,65 +267,3 @@ func TestFullConfigAndJobTriggerShareProtectedProtocolState(t *testing.T) {
|
||||
t.Fatalf("job state overwritten: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchResolvedAndDirectCreation(t *testing.T) {
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Length", "4")
|
||||
_, _ = w.Write([]byte("test"))
|
||||
}))
|
||||
t.Cleanup(upstream.Close)
|
||||
d := download.NewDownloader(&download.DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&httpProtocol.FetcherManager{}}})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { d.Close() })
|
||||
cfg, err := d.GetConfig()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.MaxRunning = 0
|
||||
if err := d.PutConfig(cfg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
service, err := NewService(d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, grouped := range []bool{false, true} {
|
||||
resolved := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", &restmodel.ResolveTask{Req: &base.Request{URL: upstream.URL + "/resolved"}})
|
||||
result := serviceTestRequest[map[string]json.RawMessage](t, service, http.MethodPost, "/api/v1/tasks/batch", &base.CreateTaskBatch{
|
||||
Group: grouped, GroupName: "folder", Opts: &base.Options{Path: t.TempDir()},
|
||||
Reqs: []*base.CreateTaskBatchItem{
|
||||
{Rid: "invalid"},
|
||||
{Rid: resolved.ID, Req: &base.Request{URL: "unsupported://ignored"}, Opts: &base.Options{SelectFiles: []int{0}}},
|
||||
{Req: &base.Request{URL: upstream.URL + "/direct"}},
|
||||
},
|
||||
})
|
||||
var ids []string
|
||||
if err := json.Unmarshal(result["taskIds"], &ids); err != nil || len(ids) != 2 {
|
||||
t.Fatalf("unexpected task IDs: %+v, %v", result, err)
|
||||
}
|
||||
wantFields := 1
|
||||
if grouped {
|
||||
wantFields = 2
|
||||
}
|
||||
if len(result) != wantFields {
|
||||
t.Fatalf("unexpected response fields: %+v", result)
|
||||
}
|
||||
var parentID string
|
||||
if grouped {
|
||||
if err := json.Unmarshal(result["parentId"], &parentID); err != nil || parentID == "" {
|
||||
t.Fatalf("missing parent ID: %+v", result)
|
||||
}
|
||||
}
|
||||
task := serviceTestRequest[*download.Task](t, service, http.MethodGet, "/api/v1/tasks/"+ids[0], nil)
|
||||
if task.ParentID != parentID || !reflect.DeepEqual(task.Meta.Opts.SelectFiles, []int{0}) {
|
||||
t.Fatalf("resolved task lost membership or selection: %+v", task)
|
||||
}
|
||||
status, response := service.JSON(http.MethodPost, "/api/v1/tasks/batch", nil, []byte(`{"reqs":[{"rid":"missing"},{"req":{"url":"unsupported://bad"}}]}`), nil)
|
||||
var failed restmodel.Result[any]
|
||||
if err := json.Unmarshal(response, &failed); err != nil || status != http.StatusOK || failed.Code == restmodel.CodeOk {
|
||||
t.Fatalf("all-failed batch succeeded: %d %s", status, response)
|
||||
}
|
||||
}
|
||||
}
|
||||
+2
-32
@@ -27,33 +27,6 @@ type Request struct {
|
||||
SkipVerifyCert bool `json:"skipVerifyCert"`
|
||||
}
|
||||
|
||||
// SetLabels replaces all labels on the request.
|
||||
//
|
||||
// Deprecated: In onResolve and onDone, assign request labels directly.
|
||||
// In onStart and onError, use task.meta.req.setLabels instead.
|
||||
func (r *Request) SetLabels(labels map[string]string) {
|
||||
r.Labels = labels
|
||||
}
|
||||
|
||||
// PutLabel sets a label on the request.
|
||||
//
|
||||
// Deprecated: In onResolve and onDone, assign request labels directly.
|
||||
// In onStart and onError, use task.meta.req.putLabel instead.
|
||||
func (r *Request) PutLabel(key, value string) {
|
||||
if r.Labels == nil {
|
||||
r.Labels = make(map[string]string)
|
||||
}
|
||||
r.Labels[key] = value
|
||||
}
|
||||
|
||||
// DelLabel deletes a label from the request.
|
||||
//
|
||||
// Deprecated: In onResolve and onDone, delete request labels directly.
|
||||
// In onStart and onError, use task.meta.req.delLabel instead.
|
||||
func (r *Request) DelLabel(key string) {
|
||||
delete(r.Labels, key)
|
||||
}
|
||||
|
||||
func (r *Request) Validate() error {
|
||||
if r.URL == "" {
|
||||
return fmt.Errorf("invalid request url")
|
||||
@@ -200,14 +173,11 @@ func ParseOptExtra[E any](opts *Options) error {
|
||||
}
|
||||
|
||||
type CreateTaskBatch struct {
|
||||
Reqs []*CreateTaskBatchItem `json:"reqs"`
|
||||
Opts *Options `json:"opts"`
|
||||
Group bool `json:"group"`
|
||||
GroupName string `json:"groupName"`
|
||||
Reqs []*CreateTaskBatchItem `json:"reqs"`
|
||||
Opts *Options `json:"opts"`
|
||||
}
|
||||
|
||||
type CreateTaskBatchItem struct {
|
||||
Rid string `json:"rid,omitempty"`
|
||||
Req *Request `json:"req"`
|
||||
Opts *Options `json:"opts"`
|
||||
}
|
||||
|
||||
@@ -32,10 +32,6 @@ func categoryForResource(cfg *base.DownloaderStoreConfig, res *base.Resource) *b
|
||||
// runs on the result, so a category directory the white list rejects falls
|
||||
// back to the base directory of this task instead of failing it.
|
||||
func (d *Downloader) decideDownloadDir(opts *base.Options, res *base.Resource) (string, error) {
|
||||
return d.decideDownloadDirForTask(opts, res, nil)
|
||||
}
|
||||
|
||||
func (d *Downloader) decideDownloadDirForTask(opts *base.Options, res *base.Resource, group *TaskGroup) (string, error) {
|
||||
if opts == nil {
|
||||
opts = &base.Options{}
|
||||
}
|
||||
@@ -52,9 +48,6 @@ func (d *Downloader) decideDownloadDirForTask(opts *base.Options, res *base.Reso
|
||||
if base == "" {
|
||||
return "", errors.New("download directory is empty")
|
||||
}
|
||||
if group != nil && opts.SkipAutoCategorize {
|
||||
return d.initDownloadPathForTask(base, group)
|
||||
}
|
||||
target := base
|
||||
if !opts.SkipAutoCategorize {
|
||||
if category := categoryForResource(cfg, res); category != nil {
|
||||
|
||||
@@ -234,38 +234,6 @@ func TestDecideDownloadDirWhiteList(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestDecideGroupDownloadDirWhiteList(t *testing.T) {
|
||||
parent := t.TempDir()
|
||||
root := filepath.Join(parent, "folder")
|
||||
nested := filepath.Join(root, "nested")
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
path string
|
||||
group *TaskGroup
|
||||
allowed string
|
||||
wantErr bool
|
||||
}{
|
||||
{name: "group root", path: root, group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*")},
|
||||
{name: "nested member", path: nested, group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*")},
|
||||
{name: "ordinary nested task", path: nested, allowed: filepath.Join(parent, "*"), wantErr: true},
|
||||
{name: "member outside group", path: filepath.Join(parent, "other"), group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*"), wantErr: true},
|
||||
{name: "group permission revoked", path: nested, group: &TaskGroup{Path: root}, allowed: nested, wantErr: true},
|
||||
{name: "literal group name", path: filepath.Join(parent, "%year%", "nested"), group: &TaskGroup{Path: filepath.Join(parent, "%year%")}, allowed: filepath.Join(parent, "*")},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
d := newUnitCategoryDownloader(newTestCategoryConfig())
|
||||
d.cfg.WhiteDownloadDirs = []string{test.allowed}
|
||||
got, err := d.decideDownloadDirForTask(&base.Options{Path: test.path, SkipAutoCategorize: true}, singleFileResource("file.txt"), test.group)
|
||||
if (err != nil) != test.wantErr {
|
||||
t.Fatalf("decideDownloadDirForTask() = %q, %v; want error = %v", got, err, test.wantErr)
|
||||
}
|
||||
if err == nil && got != test.path {
|
||||
t.Fatalf("path = %q, want %q", got, test.path)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSuggestedDownloadDir(t *testing.T) {
|
||||
programDir := filepath.Join("./downloads", "Program")
|
||||
|
||||
|
||||
+113
-512
File diff suppressed because it is too large.
Load diff
+2
-1105
File diff suppressed because it is too large.
Load diff
@@ -825,13 +825,6 @@ func newOnDoneExtensionTask(task *Task) *Task {
|
||||
return cloneExtensionTask(task)
|
||||
}
|
||||
|
||||
// SetUrl replaces the task request URL.
|
||||
//
|
||||
// Deprecated: Use task.meta.req.setUrl instead.
|
||||
func (t *ExtensionTask) SetUrl(url string) {
|
||||
t.Meta.Req.SetUrl(url)
|
||||
}
|
||||
|
||||
func (t *OnErrorExtensionTask) Continue() error {
|
||||
return t.download.Continue(&TaskFilter{
|
||||
IDs: []string{t.ID},
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
)
|
||||
|
||||
func TestExtensionDeprecatedRequestAPIs(t *testing.T) {
|
||||
for _, event := range []ActivationEvent{EventOnResolve, EventOnStart, EventOnError, EventOnDone} {
|
||||
for _, withLabels := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("%s/labels=%v", event, withLabels), func(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Clear()
|
||||
|
||||
extDir := t.TempDir()
|
||||
manifest := fmt.Sprintf(`{"name":"legacy-request","title":"Legacy request APIs","version":"0.0.1","scripts":[{"event":%q,"match":{"urls":["*://example.com/*"]},"entry":"index.js"}]}`, event)
|
||||
request := "ctx.task.meta.req"
|
||||
urlMutation := ""
|
||||
if event == EventOnResolve {
|
||||
request = "ctx.req"
|
||||
} else if event == EventOnStart || event == EventOnError {
|
||||
urlMutation = `await ctx.task.setUrl('https://example.com/legacy');
|
||||
if (req.url !== 'https://example.com/legacy') throw new Error('legacy URL did not write through');
|
||||
await req.setUrl('https://example.com/new');
|
||||
if (req.url !== 'https://example.com/new') throw new Error('new URL did not write through');
|
||||
await ctx.task.setUrl('https://example.com/final');`
|
||||
}
|
||||
script := fmt.Sprintf(`gopeed.events.%s(async ctx => {
|
||||
const req = %s;
|
||||
await req.delLabel('missing');
|
||||
await req.putLabel('initial', 'yes');
|
||||
if (req.labels.initial !== 'yes') throw new Error('putLabel did not initialize labels');
|
||||
await req.setLabels({keep: 'yes', remove: 'yes'});
|
||||
if (req.labels.initial !== undefined || req.labels.old !== undefined) throw new Error('setLabels did not replace labels');
|
||||
await req.putLabel('added', 'yes');
|
||||
await req.delLabel('remove');
|
||||
if (req.labels.keep !== 'yes' || req.labels.added !== 'yes' || req.labels.remove !== undefined) throw new Error('label mutation failed');
|
||||
%s
|
||||
await req.putLabel('completed', 'yes');
|
||||
});`, event, request, urlMutation)
|
||||
for name, content := range map[string]string{"manifest.json": manifest, "index.js": script} {
|
||||
if err := os.WriteFile(filepath.Join(extDir, name), []byte(content), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := d.InstallExtensionByFolder(extDir, false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req := &base.Request{URL: "https://example.com/original"}
|
||||
if withLabels {
|
||||
req.Labels = map[string]string{"old": "yes"}
|
||||
}
|
||||
task := &Task{Protocol: "http", Meta: &fetcher.FetcherMeta{Req: req, Opts: &base.Options{}}}
|
||||
switch event {
|
||||
case EventOnResolve:
|
||||
if _, err := d.triggerOnResolve(req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case EventOnStart:
|
||||
d.triggerOnStart(task)
|
||||
case EventOnError:
|
||||
d.triggerOnError(task, errors.New("retry"))
|
||||
case EventOnDone:
|
||||
d.triggerOnDone(task)
|
||||
}
|
||||
if !reflect.DeepEqual(req.Labels, map[string]string{"keep": "yes", "added": "yes", "completed": "yes"}) {
|
||||
t.Fatalf("legacy handler did not update the original request: %#v", req.Labels)
|
||||
}
|
||||
expectedURL := "https://example.com/original"
|
||||
if urlMutation != "" {
|
||||
expectedURL = "https://example.com/final"
|
||||
}
|
||||
if req.URL != expectedURL {
|
||||
t.Fatalf("URL = %q, want %q", req.URL, expectedURL)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -152,7 +152,7 @@ func TestExtensionOnErrorRequestWritesToOriginalTask(t *testing.T) {
|
||||
request.setHeaders({Keep:'new'});
|
||||
request.putHeader('Authorization','Bearer new');
|
||||
request.setMethod('POST');
|
||||
if (typeof task.setUrl !== 'function' || typeof task.extra !== 'undefined') throw new Error('unexpected legacy task API');
|
||||
if (typeof task.setUrl !== 'undefined' || typeof task.extra !== 'undefined') throw new Error('legacy task API');
|
||||
request.setUrl('https://example.com/new');
|
||||
request.setLabels({keep:'yes',remove:'yes'});
|
||||
request.putLabel('new','value');
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
)
|
||||
|
||||
func TestDownloader_Extension_MagnetMatch(t *testing.T) {
|
||||
const magnetURL = "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567"
|
||||
for _, pattern := range []string{"magnet:*", "magnet:?*"} {
|
||||
t.Run(pattern, func(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{
|
||||
Storage: NewMemStorage(),
|
||||
StorageDir: t.TempDir(),
|
||||
TempDir: t.TempDir(),
|
||||
})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Clear()
|
||||
|
||||
extDir := t.TempDir()
|
||||
manifest := fmt.Sprintf(`{"name":"magnet-match","author":"test","title":"Magnet match","version":"1.0.0","scripts":[{"event":"onResolve","match":{"urls":[%q]},"entry":"index.js"}]}`, pattern)
|
||||
script := `gopeed.events.onResolve(ctx => {
|
||||
ctx.res = {files: [{name: "resolved.txt", size: 1, req: {url: "https://example.com/resolved.txt"}}]};
|
||||
});`
|
||||
for name, content := range map[string]string{"manifest.json": manifest, "index.js": script} {
|
||||
if err := os.WriteFile(filepath.Join(extDir, name), []byte(content), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := d.InstallExtensionByFolder(extDir, true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
for _, req := range []*base.Request{
|
||||
{URL: magnetURL},
|
||||
{URL: "https://example.com/rewritten", RawURL: magnetURL},
|
||||
} {
|
||||
// Exercise extension activation without falling back to a live
|
||||
// BitTorrent metadata lookup when the pattern fails to match.
|
||||
res, err := d.triggerOnResolve(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res == nil || len(res.Files) != 1 || res.Files[0].Name != "resolved.txt" {
|
||||
t.Fatalf("magnet extension did not resolve request %+v: %+v", req, res)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -70,11 +70,10 @@ func TestExtensionResolveSizeThroughTaskCreation(t *testing.T) {
|
||||
if err := json.Unmarshal(wire, &submitted); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
batchResult, err := d.CreateDirectBatch(&submitted)
|
||||
ids, err := d.CreateDirectBatch(&submitted)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := batchResult
|
||||
for i, id := range ids {
|
||||
task := d.GetTask(id)
|
||||
if task.Meta.Res != nil {
|
||||
|
||||
@@ -2044,13 +2044,8 @@ func waitForTaskStatus(t *testing.T, downloader *Downloader, id string, status b
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
task := downloader.GetTask(id)
|
||||
if task != nil && downloader.taskStatus(task) == status {
|
||||
if status != base.DownloadStatusDone {
|
||||
return
|
||||
}
|
||||
if _, watching := downloader.watchedTasks.Load(id); !watching {
|
||||
return
|
||||
}
|
||||
if task != nil && task.Status == status {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
@@ -2058,6 +2053,9 @@ func waitForTaskStatus(t *testing.T, downloader *Downloader, id string, status b
|
||||
if task == nil {
|
||||
t.Fatalf("timeout waiting for task %s status %s: task not found", id, status)
|
||||
}
|
||||
snapshot, err := downloader.RuntimeStatus(id)
|
||||
t.Fatalf("timeout waiting for task %s status %s: got %+v, error=%v", id, status, snapshot, err)
|
||||
total := int64(0)
|
||||
if task.Meta != nil && task.Meta.Res != nil {
|
||||
total = task.Meta.Res.Size
|
||||
}
|
||||
t.Fatalf("timeout waiting for task %s status %s: got %s downloaded=%d total=%d", id, status, task.Status, task.Progress.Downloaded, total)
|
||||
}
|
||||
@@ -1,57 +0,0 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
"github.com/GopeedLab/gopeed/pkg/protocol/ftp"
|
||||
"github.com/GopeedLab/gopeed/pkg/protocol/http"
|
||||
)
|
||||
|
||||
// TestTorrentNameIsFile checks the default manager order: an FTP URL whose
|
||||
// name looks like a torrent or a playlist is still an FTP download.
|
||||
func TestTorrentNameIsFile(t *testing.T) {
|
||||
d := NewDownloader(nil)
|
||||
for u, want := range map[string]string{
|
||||
"ftp://h/x.torrent": "ftp",
|
||||
"ftp://h/v.m3u8": "ftp",
|
||||
"ftps://h/x.torrent": "ftp",
|
||||
"ftpes://h/a.bin": "ftp",
|
||||
"https://h/v.m3u8": "hls",
|
||||
"https://h/x.torrent": "http",
|
||||
"magnet:?xt=urn:btih:": "bt",
|
||||
} {
|
||||
fm, err := d.parseFm(u)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", u, err)
|
||||
}
|
||||
if fm.Name() != want {
|
||||
t.Fatalf("%s goes to %q, want %q", u, fm.Name(), want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostDownloadExtraAcceptsFTP(t *testing.T) {
|
||||
yes := true
|
||||
file := &base.Resource{Files: []*base.FileInfo{{Name: "a.zip"}}}
|
||||
folder := &base.Resource{Name: "dir", Files: []*base.FileInfo{{Name: "a.zip"}}}
|
||||
|
||||
httpExtra := &http.OptsExtra{AutoExtract: &yes}
|
||||
if got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{Extra: httpExtra}}); got != httpExtra {
|
||||
t.Fatalf("HTTP options must pass through unchanged, got %+v", got)
|
||||
}
|
||||
|
||||
ftpExtra := &ftp.OptsExtra{Connections: 4, AutoTorrent: &yes, DeleteTorrentAfterDownload: &yes, AutoExtract: &yes, ArchivePassword: "pw", DeleteAfterExtract: true}
|
||||
got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{Extra: ftpExtra}})
|
||||
if got == nil || got.AutoTorrent != &yes || got.DeleteTorrentAfterDownload != &yes || got.AutoExtract != &yes ||
|
||||
got.ArchivePassword != "pw" || !got.DeleteAfterExtract {
|
||||
t.Fatalf("FTP options were not carried over: %+v", got)
|
||||
}
|
||||
if got := postDownloadExtra(&fetcher.FetcherMeta{Res: folder, Opts: &base.Options{Extra: ftpExtra}}); got != nil {
|
||||
t.Fatalf("an FTP folder has no single file to extract, got %+v", got)
|
||||
}
|
||||
if got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{}}); got != nil {
|
||||
t.Fatalf("no options, got %+v", got)
|
||||
}
|
||||
}
|
||||
+9
-34
@@ -10,7 +10,6 @@ import (
|
||||
"github.com/GopeedLab/gopeed/internal/production"
|
||||
"github.com/GopeedLab/gopeed/internal/protocol/bt"
|
||||
"github.com/GopeedLab/gopeed/internal/protocol/ed2k"
|
||||
"github.com/GopeedLab/gopeed/internal/protocol/ftp"
|
||||
"github.com/GopeedLab/gopeed/internal/protocol/hls"
|
||||
"github.com/GopeedLab/gopeed/internal/protocol/http"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
@@ -29,34 +28,18 @@ type ResolveResult struct {
|
||||
SuggestedDownloadDir string `json:"suggestedDownloadDir,omitempty"`
|
||||
}
|
||||
|
||||
// TaskGroup stores the directory shared by independently scheduled tasks.
|
||||
type TaskGroup struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Path string `json:"path"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
}
|
||||
|
||||
// CreateTaskBatchResult is shared by ordinary and grouped batches.
|
||||
type CreateTaskBatchResult struct {
|
||||
TaskIDs []string `json:"taskIds"`
|
||||
ParentID string `json:"parentId,omitempty"`
|
||||
}
|
||||
|
||||
// Task represents a download unit.
|
||||
// Options flow path:
|
||||
// REST Request (model.CreateTask/ResolveTask) -> pkg/download Downloader -> Task.Meta.Opts (base.Options) -> Fetcher.Resolve/Start (e.g. internal/protocol/http/fetcher.go)
|
||||
type Task struct {
|
||||
ID string `json:"id"`
|
||||
Protocol string `json:"protocol"`
|
||||
Meta *fetcher.FetcherMeta `json:"meta"`
|
||||
// ParentID identifies the independently stored group.
|
||||
ParentID string `json:"parentId,omitempty"`
|
||||
Status base.Status `json:"status"`
|
||||
Uploading bool `json:"uploading"`
|
||||
Progress *Progress `json:"progress"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
ID string `json:"id"`
|
||||
Protocol string `json:"protocol"`
|
||||
Meta *fetcher.FetcherMeta `json:"meta"`
|
||||
Status base.Status `json:"status"`
|
||||
Uploading bool `json:"uploading"`
|
||||
Progress *Progress `json:"progress"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
// DoneAt is the time when the task completed; nil if it has never
|
||||
// finished. A done task never leaves the done state, so the value is
|
||||
// written once and kept.
|
||||
@@ -70,8 +53,6 @@ type Task struct {
|
||||
// does not have it and keeps its directory and name.
|
||||
NeedsPreparation bool `json:"needsPreparation,omitempty"`
|
||||
|
||||
group *TaskGroup
|
||||
|
||||
producer production.Source
|
||||
producerReceived int64
|
||||
|
||||
@@ -162,12 +143,10 @@ func (t *Task) MarshalJSON() ([]byte, error) {
|
||||
type rawTaskType Task
|
||||
jsonTask := struct {
|
||||
*rawTaskType
|
||||
Name string `json:"name"`
|
||||
Group *TaskGroup `json:"group,omitempty"`
|
||||
Name string `json:"name"`
|
||||
}{
|
||||
(*rawTaskType)(t),
|
||||
t.Name(),
|
||||
t.group,
|
||||
}
|
||||
return json.Marshal(jsonTask)
|
||||
}
|
||||
@@ -250,10 +229,6 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
|
||||
}
|
||||
if len(cfg.FetchManagers) == 0 {
|
||||
cfg.FetchManagers = []fetcher.FetcherManager{
|
||||
// FTP matches by URL scheme only. It comes before the managers that
|
||||
// match by file extension, so ftp://host/x.torrent or x.m3u8 is
|
||||
// downloaded as a file over FTP.
|
||||
new(ftp.FetcherManager),
|
||||
new(hls.FetcherManager),
|
||||
new(http.FetcherManager),
|
||||
new(bt.FetcherManager),
|
||||
|
||||
@@ -27,10 +27,6 @@ type ScriptData struct {
|
||||
// ScriptPayload contains the task data
|
||||
type ScriptPayload struct {
|
||||
Task *Task
|
||||
// TaskName is computed before Task is cloned. The clone has no fetcher
|
||||
// manager, so for an unresolved task without Options.Name its Name()
|
||||
// would return "unknown".
|
||||
TaskName string
|
||||
}
|
||||
|
||||
// getScriptPaths extracts script paths from config
|
||||
@@ -107,7 +103,7 @@ func (d *Downloader) executeScriptAtPath(scriptPath string, data *ScriptData) er
|
||||
cmd.Env = append(os.Environ(),
|
||||
fmt.Sprintf("GOPEED_EVENT=%s", data.Event),
|
||||
fmt.Sprintf("GOPEED_TASK_ID=%s", data.Payload.Task.ID),
|
||||
fmt.Sprintf("GOPEED_TASK_NAME=%s", data.Payload.TaskName),
|
||||
fmt.Sprintf("GOPEED_TASK_NAME=%s", data.Payload.Task.Name()),
|
||||
fmt.Sprintf("GOPEED_TASK_STATUS=%s", data.Payload.Task.Status),
|
||||
)
|
||||
|
||||
@@ -141,8 +137,7 @@ func (d *Downloader) triggerScripts(event ScriptEvent, task *Task, err error) {
|
||||
Event: event,
|
||||
Time: time.Now().UnixMilli(),
|
||||
Payload: &ScriptPayload{
|
||||
Task: task.clone(),
|
||||
TaskName: task.Name(),
|
||||
Task: task.clone(),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
Loaded 50 of 175 files, more files were not shown because too many files have changed in this diff.
Show more
Reference in new issue
Block a user