From 38ff711ea95e6237c113178fbb9cf85d24c2612e Mon Sep 17 00:00:00 2001 From: Levi Date: Fri, 19 Jan 2024 18:04:54 +0800 Subject: [PATCH] feat: add batch delete tasks api (#338) --- pkg/download/downloader.go | 64 ++++++++++++++++++++++++++++++++++++++ pkg/rest/api.go | 35 ++++++++++++--------- pkg/rest/server.go | 1 + pkg/rest/server_test.go | 52 +++++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 14 deletions(-) diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 766d2859..6c87557c 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -448,6 +448,13 @@ func (d *Downloader) Delete(id string, force bool) (err error) { for i, t := range d.tasks { if t.ID == id { d.tasks = append(d.tasks[:i], d.tasks[i+1:]...) + break + } + } + for i, t := range d.waitTasks { + if t.ID == id { + d.waitTasks = append(d.waitTasks[:i], d.waitTasks[i+1:]...) + break } } }() @@ -460,6 +467,47 @@ func (d *Downloader) Delete(id string, force bool) (err error) { return } +func (d *Downloader) DeleteByStatues(statues []base.Status, force bool) (err error) { + deleteTasks := d.GetTasksByStatues(statues) + if len(deleteTasks) == 0 { + return + } + + deleteIds := make([]string, 0) + for _, task := range deleteTasks { + deleteIds = append(deleteIds, task.ID) + } + func() { + d.lock.Lock() + defer d.lock.Unlock() + + for _, id := range deleteIds { + for i, t := range d.tasks { + if t.ID == id { + d.tasks = append(d.tasks[:i], d.tasks[i+1:]...) + break + } + } + for i, t := range d.waitTasks { + if t.ID == id { + d.waitTasks = append(d.waitTasks[:i], d.waitTasks[i+1:]...) + break + } + } + } + }() + + for _, task := range deleteTasks { + err = d.doDelete(task, force) + if err != nil { + return + } + } + + d.notifyRunning() + return +} + func (d *Downloader) doDelete(task *Task, force bool) (err error) { err = func() error { if task.fetcher != nil { @@ -548,6 +596,22 @@ func (d *Downloader) GetTasks() []*Task { return d.tasks } +func (d *Downloader) GetTasksByStatues(statues []base.Status) []*Task { + if len(statues) == 0 { + return d.tasks + } + tasks := make([]*Task, 0) + for _, task := range d.tasks { + for _, status := range statues { + if task.Status == status { + tasks = append(tasks, task) + break + } + } + } + return tasks +} + func (d *Downloader) GetConfig() (*DownloaderStoreConfig, error) { return d.cfg.DownloaderStoreConfig, nil } diff --git a/pkg/rest/api.go b/pkg/rest/api.go index a0fc3ea8..79378254 100644 --- a/pkg/rest/api.go +++ b/pkg/rest/api.go @@ -104,6 +104,17 @@ func DeleteTask(w http.ResponseWriter, r *http.Request) { WriteJson(w, model.NewNilResult()) } +func DeleteTasks(w http.ResponseWriter, r *http.Request) { + r.ParseForm() + status := r.Form["status"] + force := r.FormValue("force") + if err := Downloader.DeleteByStatues(convertStatues(status), force == "true"); err != nil { + WriteJson(w, model.NewErrorResult(err.Error())) + return + } + WriteJson(w, model.NewNilResult()) +} + func GetTask(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) taskId := vars["id"] @@ -122,20 +133,8 @@ func GetTask(w http.ResponseWriter, r *http.Request) { func GetTasks(w http.ResponseWriter, r *http.Request) { r.ParseForm() status := r.Form["status"] - tasks := Downloader.GetTasks() - if len(status) == 0 { - WriteJson(w, model.NewOkResult(tasks)) - return - } - result := make([]*download.Task, 0) - for _, task := range tasks { - for _, s := range status { - if task.Status == base.Status(s) { - result = append(result, task) - } - } - } - WriteJson(w, model.NewOkResult(result)) + tasks := Downloader.GetTasksByStatues(convertStatues(status)) + WriteJson(w, model.NewOkResult(tasks)) } func GetConfig(w http.ResponseWriter, r *http.Request) { @@ -282,6 +281,14 @@ func DoProxy(w http.ResponseWriter, r *http.Request) { w.Write(buf) } +func convertStatues(statues []string) []base.Status { + result := make([]base.Status, 0) + for _, status := range statues { + result = append(result, base.Status(status)) + } + return result +} + func writeError(w http.ResponseWriter, msg string) { w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(msg)) diff --git a/pkg/rest/server.go b/pkg/rest/server.go index 9f199027..73fce377 100644 --- a/pkg/rest/server.go +++ b/pkg/rest/server.go @@ -95,6 +95,7 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error r.Methods(http.MethodPut).Path("/api/v1/tasks/pause").HandlerFunc(PauseAllTask) r.Methods(http.MethodPut).Path("/api/v1/tasks/continue").HandlerFunc(ContinueAllTask) r.Methods(http.MethodDelete).Path("/api/v1/tasks/{id}").HandlerFunc(DeleteTask) + r.Methods(http.MethodDelete).Path("/api/v1/tasks").HandlerFunc(DeleteTasks) r.Methods(http.MethodGet).Path("/api/v1/tasks/{id}").HandlerFunc(GetTask) r.Methods(http.MethodGet).Path("/api/v1/tasks").HandlerFunc(GetTasks) r.Methods(http.MethodGet).Path("/api/v1/config").HandlerFunc(GetConfig) diff --git a/pkg/rest/server_test.go b/pkg/rest/server_test.go index 676b6fd4..0b38d183 100644 --- a/pkg/rest/server_test.go +++ b/pkg/rest/server_test.go @@ -242,6 +242,58 @@ func TestDeleteTaskForce(t *testing.T) { }) } +func TestDeleteAllTasks(t *testing.T) { + doTest(func() { + taskCount := 3 + + var wg sync.WaitGroup + wg.Add(taskCount) + Downloader.Listener(func(event *download.Event) { + if event.Key == download.EventKeyFinally { + wg.Done() + } + }) + + for i := 0; i < taskCount; i++ { + httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + } + + wg.Wait() + + httpRequestCheckOk[any](http.MethodDelete, "/api/v1/tasks?force=true", nil) + tasks := httpRequestCheckOk[[]*download.Task](http.MethodGet, "/api/v1/tasks", nil) + if len(tasks) != 0 { + t.Errorf("DeleteTasks() got = %v, want %v", len(tasks), 0) + } + }) +} + +func TestDeleteTasksByStatues(t *testing.T) { + doTest(func() { + taskCount := 3 + + var wg sync.WaitGroup + wg.Add(taskCount) + Downloader.Listener(func(event *download.Event) { + if event.Key == download.EventKeyFinally { + wg.Done() + } + }) + + for i := 0; i < taskCount; i++ { + httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + } + + wg.Wait() + + httpRequestCheckOk[any](http.MethodDelete, fmt.Sprintf("/api/v1/tasks?status=%s&force=true", base.DownloadStatusDone), nil) + tasks := httpRequestCheckOk[[]*download.Task](http.MethodGet, "/api/v1/tasks", nil) + if len(tasks) != 0 { + t.Errorf("DeleteTasks() got = %v, want %v", len(tasks), 0) + } + }) +} + func TestGetTasks(t *testing.T) { doTest(func() { var wg sync.WaitGroup