mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-07 17:10:29 +00:00
feat: implement task patching functionality (#1267)
This commit is contained in:
22 files changed
+1281
-132
No files matched your search
+2
-2
@@ -78,7 +78,7 @@ func (r *Resource) Validate() error {
|
||||
if r.Name == "" {
|
||||
return fmt.Errorf("invalid resource name")
|
||||
}
|
||||
if r.Files == nil || len(r.Files) == 0 {
|
||||
if len(r.Files) == 0 {
|
||||
return fmt.Errorf("invalid resource files")
|
||||
}
|
||||
for _, file := range r.Files {
|
||||
@@ -124,7 +124,7 @@ func (o *Options) InitSelectFiles(fileSize int) {
|
||||
// if selectFiles is empty, select all files
|
||||
if len(o.SelectFiles) == 0 {
|
||||
o.SelectFiles = make([]int, fileSize)
|
||||
for i := 0; i < fileSize; i++ {
|
||||
for i := range fileSize {
|
||||
o.SelectFiles[i] = i
|
||||
}
|
||||
}
|
||||
|
||||
@@ -480,6 +480,47 @@ func (d *Downloader) Create(rrId string) (taskId string, err error) {
|
||||
return d.doCreate(fetcher, nil)
|
||||
}
|
||||
|
||||
// Patch modifies task-specific data based on the protocol.
|
||||
// For HTTP protocol, it can modify Request info.
|
||||
// For BT protocol, it can modify SelectFiles.
|
||||
func (d *Downloader) Patch(id string, req *base.Request, opts *base.Options) error {
|
||||
task := d.GetTask(id)
|
||||
if task == nil {
|
||||
return ErrTaskNotFound
|
||||
}
|
||||
|
||||
// Restore fetcher if not loaded
|
||||
if task.fetcher == nil {
|
||||
err := func() error {
|
||||
task.statusLock.Lock()
|
||||
defer task.statusLock.Unlock()
|
||||
|
||||
return d.restoreFetcher(task)
|
||||
}()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Call the fetcher's Patch method
|
||||
if err := task.fetcher.Patch(req, opts); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Update task meta from fetcher
|
||||
task.Meta = task.fetcher.Meta()
|
||||
|
||||
// Save task to storage
|
||||
if err := d.saveTask(task); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Emit progress event to notify listeners
|
||||
d.emit(EventKeyProgress, task)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Downloader) Pause(filter *TaskFilter) (err error) {
|
||||
if filter == nil || filter.IsEmpty() {
|
||||
return d.pauseAll()
|
||||
|
||||
@@ -2648,3 +2648,87 @@ func TestDownloader_AutoTorrentDisabled(t *testing.T) {
|
||||
t.Error("Torrent file should have been downloaded")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDownloader_PatchTask_HTTP(t *testing.T) {
|
||||
listener := test.StartTestFileServer()
|
||||
defer listener.Close()
|
||||
|
||||
downloader := NewDownloader(nil)
|
||||
if err := downloader.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer downloader.Clear()
|
||||
|
||||
req := &base.Request{
|
||||
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
|
||||
Extra: &http.OptsExtra{
|
||||
Connections: 2,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"test": "value1",
|
||||
},
|
||||
}
|
||||
opts := &base.Options{
|
||||
Path: test.Dir,
|
||||
Name: test.DownloadName,
|
||||
}
|
||||
|
||||
// Create task but don't start it yet
|
||||
taskId, err := downloader.CreateDirect(req, opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Pause the task immediately
|
||||
if err := downloader.Pause(&TaskFilter{IDs: []string{taskId}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Patch the task with new labels
|
||||
patchReq := &base.Request{
|
||||
Labels: map[string]string{
|
||||
"test": "value2",
|
||||
"newKey": "newValue",
|
||||
},
|
||||
}
|
||||
|
||||
if err := downloader.Patch(taskId, patchReq, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify the patch was applied
|
||||
task := downloader.GetTask(taskId)
|
||||
if task == nil {
|
||||
t.Fatal("task not found")
|
||||
}
|
||||
|
||||
if task.Meta.Req.Labels["test"] != "value2" {
|
||||
t.Errorf("PatchTask() label 'test' = %v, want %v", task.Meta.Req.Labels["test"], "value2")
|
||||
}
|
||||
if task.Meta.Req.Labels["newKey"] != "newValue" {
|
||||
t.Errorf("PatchTask() label 'newKey' = %v, want %v", task.Meta.Req.Labels["newKey"], "newValue")
|
||||
}
|
||||
|
||||
// Clean up
|
||||
downloader.Delete(&TaskFilter{IDs: []string{taskId}}, true)
|
||||
}
|
||||
|
||||
func TestDownloader_PatchTask_NotFound(t *testing.T) {
|
||||
downloader := NewDownloader(nil)
|
||||
if err := downloader.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer downloader.Clear()
|
||||
|
||||
// Try to patch a non-existent task
|
||||
patchReq := &base.Request{
|
||||
Labels: map[string]string{
|
||||
"test": "value",
|
||||
},
|
||||
}
|
||||
|
||||
err := downloader.Patch("non-existent-id", patchReq, nil)
|
||||
if err != ErrTaskNotFound {
|
||||
t.Errorf("Patch() error = %v, want %v", err, ErrTaskNotFound)
|
||||
}
|
||||
}
|
||||
@@ -74,6 +74,28 @@ func CreateTaskBatch(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
func PatchTask(w http.ResponseWriter, r *http.Request) {
|
||||
vars := mux.Vars(r)
|
||||
taskId := vars["id"]
|
||||
if taskId == "" {
|
||||
WriteJson(w, model.NewErrorResult("param invalid: id", model.CodeInvalidParam))
|
||||
return
|
||||
}
|
||||
|
||||
var req model.ResolveTask
|
||||
if ReadJson(r, w, &req) {
|
||||
if err := Downloader.Patch(taskId, req.Req, req.Opts); err != nil {
|
||||
if err == download.ErrTaskNotFound {
|
||||
WriteJson(w, model.NewErrorResult("task not found", model.CodeTaskNotFound))
|
||||
return
|
||||
}
|
||||
WriteJson(w, model.NewErrorResult(err.Error()))
|
||||
return
|
||||
}
|
||||
WriteJson(w, model.NewNilResult())
|
||||
}
|
||||
}
|
||||
|
||||
func PauseTask(w http.ResponseWriter, r *http.Request) {
|
||||
filter, errResult := parseIdFilter(r)
|
||||
if errResult != nil {
|
||||
|
||||
@@ -6,6 +6,7 @@ type ResolveTask struct {
|
||||
Req *base.Request `json:"req"`
|
||||
Opts *base.Options `json:"opts"`
|
||||
}
|
||||
|
||||
type CreateTask struct {
|
||||
Rid string `json:"rid"`
|
||||
|
||||
|
||||
+12
-10
@@ -8,12 +8,6 @@ import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/GopeedLab/gopeed/pkg/download"
|
||||
"github.com/GopeedLab/gopeed/pkg/rest/model"
|
||||
"github.com/GopeedLab/gopeed/pkg/util"
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/pkg/errors"
|
||||
"io"
|
||||
"io/fs"
|
||||
"net"
|
||||
@@ -24,6 +18,13 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/download"
|
||||
"github.com/GopeedLab/gopeed/pkg/rest/model"
|
||||
"github.com/GopeedLab/gopeed/pkg/util"
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -122,6 +123,7 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error
|
||||
r.Methods(http.MethodPost).Path("/api/v1/resolve").HandlerFunc(Resolve)
|
||||
r.Methods(http.MethodPost).Path("/api/v1/tasks").HandlerFunc(CreateTask)
|
||||
r.Methods(http.MethodPost).Path("/api/v1/tasks/batch").HandlerFunc(CreateTaskBatch)
|
||||
r.Methods(http.MethodPatch).Path("/api/v1/tasks/{id}").HandlerFunc(PatchTask)
|
||||
r.Methods(http.MethodPut).Path("/api/v1/tasks/{id}/pause").HandlerFunc(PauseTask)
|
||||
r.Methods(http.MethodPut).Path("/api/v1/tasks/pause").HandlerFunc(PauseTasks)
|
||||
r.Methods(http.MethodPut).Path("/api/v1/tasks/{id}/continue").HandlerFunc(ContinueTask)
|
||||
@@ -145,9 +147,9 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error
|
||||
r.Path("/api/v1/proxy").HandlerFunc(DoProxy)
|
||||
|
||||
enableApiToken := startCfg.ApiToken != ""
|
||||
enableBasicAuth := startCfg.WebEnable && startCfg.WebAuth != nil
|
||||
enableWebAuth := startCfg.WebEnable && startCfg.WebAuth != nil
|
||||
if startCfg.WebEnable {
|
||||
if enableBasicAuth {
|
||||
if enableWebAuth {
|
||||
r.Methods(http.MethodPost).Path("/api/web/login").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var loginReq model.WebAuth
|
||||
if ReadJson(r, w, &loginReq) {
|
||||
@@ -172,7 +174,7 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error
|
||||
r.PathPrefix("/fs/extensions").Handler(http.FileServer(new(extensionFileSystem)))
|
||||
r.PathPrefix("/").Handler(gzipMiddleware(http.FileServer(newEmbedCacheFileSystem(http.FS(startCfg.WebFS)))))
|
||||
}
|
||||
if enableApiToken || enableBasicAuth {
|
||||
if enableApiToken || enableWebAuth {
|
||||
writeUnauthorized := func(w http.ResponseWriter, r *http.Request) {
|
||||
WriteStatusJson(w, http.StatusUnauthorized, model.NewErrorResult("unauthorized", model.CodeUnauthorized))
|
||||
}
|
||||
@@ -193,7 +195,7 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error
|
||||
}
|
||||
}
|
||||
|
||||
if enableBasicAuth {
|
||||
if enableWebAuth {
|
||||
if !strings.HasPrefix(r.URL.Path, "/api/") || r.URL.Path == "/api/web/login" {
|
||||
h.ServeHTTP(w, r)
|
||||
return
|
||||
|
||||
@@ -232,6 +232,60 @@ func TestPauseAndContinueTask(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestPatchTask(t *testing.T) {
|
||||
doTest(func() {
|
||||
// Create a task
|
||||
taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq)
|
||||
if taskId == "" {
|
||||
t.Fatal("create task failed")
|
||||
}
|
||||
|
||||
// Pause the task
|
||||
httpRequestCheckOk[any](http.MethodPut, "/api/v1/tasks/"+taskId+"/pause", nil)
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
|
||||
// Patch the task with new labels
|
||||
patchReq := &model.ResolveTask{
|
||||
Req: &base.Request{
|
||||
Labels: map[string]string{
|
||||
"patched": "true",
|
||||
"version": "2",
|
||||
},
|
||||
},
|
||||
}
|
||||
httpRequestCheckOk[any](http.MethodPatch, "/api/v1/tasks/"+taskId, patchReq)
|
||||
|
||||
// Verify the patch was applied
|
||||
task := httpRequestCheckOk[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil)
|
||||
if task.Meta.Req.Labels["patched"] != "true" {
|
||||
t.Errorf("PatchTask() label 'patched' = %v, want %v", task.Meta.Req.Labels["patched"], "true")
|
||||
}
|
||||
if task.Meta.Req.Labels["version"] != "2" {
|
||||
t.Errorf("PatchTask() label 'version' = %v, want %v", task.Meta.Req.Labels["version"], "2")
|
||||
}
|
||||
|
||||
// Clean up
|
||||
httpRequestCheckOk[any](http.MethodDelete, "/api/v1/tasks/"+taskId+"?force=true", nil)
|
||||
})
|
||||
}
|
||||
|
||||
func TestPatchTaskNotFound(t *testing.T) {
|
||||
doTest(func() {
|
||||
// Try to patch a non-existent task
|
||||
patchReq := &model.ResolveTask{
|
||||
Req: &base.Request{
|
||||
Labels: map[string]string{
|
||||
"test": "value",
|
||||
},
|
||||
},
|
||||
}
|
||||
code, _ := httpRequest[any](http.MethodPatch, "/api/v1/tasks/non-existent-id", patchReq)
|
||||
if code != int(model.CodeTaskNotFound) {
|
||||
t.Errorf("PatchTaskNotFound() result code = %v, want %v", code, model.CodeTaskNotFound)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestPauseAllAndContinueALLTasks(t *testing.T) {
|
||||
doTest(func() {
|
||||
cfg, err := Downloader.GetConfig()
|
||||
|
||||
Reference in new issue
Block a user