feat: use batch download api for create multitasks (#978)

pkg/base/model.go

@@ -163,6 +163,16 @@ func ParseOptsExtra[E any](opts *Options) error {
 	return nil
 }
 
+type CreateTaskBatch struct {
+	Reqs []*CreateTaskBatchItem `json:"reqs"`
+	Opts *Options               `json:"opts"`
+}
+
+type CreateTaskBatchItem struct {
+	Req  *Request `json:"req"`
+	Opts *Options `json:"opts"`
+}
+
 // DownloaderStoreConfig is the config that can restore the downloader.
 type DownloaderStoreConfig struct {
 	FirstLoad bool `json:"-"` // FirstLoad is the flag that the config is first time init and not from store

pkg/download/downloader.go

@@ -359,10 +359,14 @@ func (d *Downloader) CreateDirect(req *base.Request, opts *base.Options) (taskId
 	return d.doCreate(fetcher, opts)
 }
 
-func (d *Downloader) CreateDirectBatch(reqs []*base.Request, opts *base.Options) (taskId []string, err error) {
+func (d *Downloader) CreateDirectBatch(req *base.CreateTaskBatch) (taskId []string, err error) {
 	taskIds := make([]string, 0)
-	for _, req := range reqs {
-		taskId, err := d.CreateDirect(req, opts.Clone())
+	for _, ir := range req.Reqs {
+		opts := ir.Opts
+		if opts == nil {
+			opts = req.Opts
+		}
+		taskId, err := d.CreateDirect(ir.Req, opts.Clone())
 		if err != nil {
 			return nil, err
 		}
@@ -574,15 +578,19 @@ func (d *Downloader) Delete(filter *TaskFilter, force bool) (err error) {
 }
 
 func (d *Downloader) deleteAll() (err error) {
+	var deleteTasksTemp []*Task
 	func() {
 		d.lock.Lock()
 		defer d.lock.Unlock()
 
+		for _, task := range d.tasks {
+			deleteTasksTemp = append(deleteTasksTemp, task)
+		}
 		d.tasks = make([]*Task, 0)
 		d.waitTasks = make([]*Task, 0)
 	}()
 
-	for _, task := range d.tasks {
+	for _, task := range deleteTasksTemp {
 		if err = d.doDelete(task, true); err != nil {
 			return
 		}

pkg/download/downloader_test.go

@@ -137,13 +137,15 @@ func TestDownloader_CreateDirectBatch(t *testing.T) {
 		downloader.Clear()
 	}()
 
-	reqs := make([]*base.Request, 0)
+	reqs := make([]*base.CreateTaskBatchItem, 0)
 	fileNames := make([]string, 0)
 	for i := 0; i < 5; i++ {
 		req := &base.Request{
 			URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
 		}
-		reqs = append(reqs, req)
+		reqs = append(reqs, &base.CreateTaskBatchItem{
+			Req: req,
+		})
 		if i == 0 {
 			fileNames = append(fileNames, test.DownloadName)
 		} else {
@@ -160,11 +162,14 @@ func TestDownloader_CreateDirectBatch(t *testing.T) {
 		}
 	})
 
-	_, err := downloader.CreateDirectBatch(reqs, &base.Options{
-		Path: test.Dir,
-		Name: test.DownloadName,
-		Extra: http.OptsExtra{
-			Connections: 4,
+	_, err := downloader.CreateDirectBatch(&base.CreateTaskBatch{
+		Reqs: reqs,
+		Opts: &base.Options{
+			Path: test.Dir,
+			Name: test.DownloadName,
+			Extra: http.OptsExtra{
+				Connections: 4,
+			},
 		},
 	})
 	if err != nil {
@@ -499,13 +504,15 @@ func TestDownloader_GetTasksByFilter(t *testing.T) {
 		downloader.Clear()
 	}()
 
-	reqs := make([]*base.Request, 0)
+	reqs := make([]*base.CreateTaskBatchItem, 0)
 	fileNames := make([]string, 0)
 	for i := 0; i < 10; i++ {
 		req := &base.Request{
 			URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
 		}
-		reqs = append(reqs, req)
+		reqs = append(reqs, &base.CreateTaskBatchItem{
+			Req: req,
+		})
 		if i == 0 {
 			fileNames = append(fileNames, test.DownloadName)
 		} else {
@@ -522,11 +529,14 @@ func TestDownloader_GetTasksByFilter(t *testing.T) {
 		}
 	})
 
-	taskIds, err := downloader.CreateDirectBatch(reqs, &base.Options{
-		Path: test.Dir,
-		Name: test.DownloadName,
-		Extra: http.OptsExtra{
-			Connections: 4,
+	taskIds, err := downloader.CreateDirectBatch(&base.CreateTaskBatch{
+		Reqs: reqs,
+		Opts: &base.Options{
+			Path: test.Dir,
+			Name: test.DownloadName,
+			Extra: http.OptsExtra{
+				Connections: 4,
+			},
 		},
 	})
 	if err != nil {

pkg/rest/api.go

@@ -58,13 +58,13 @@ func CreateTask(w http.ResponseWriter, r *http.Request) {
 }
 
 func CreateTaskBatch(w http.ResponseWriter, r *http.Request) {
-	var req model.CreateTaskBatch
+	var req base.CreateTaskBatch
 	if ReadJson(r, w, &req) {
 		if len(req.Reqs) == 0 {
 			WriteJson(w, model.NewErrorResult("param invalid: reqs", model.CodeInvalidParam))
 			return
 		}
-		taskIds, err := Downloader.CreateDirectBatch(req.Reqs, req.Opt)
+		taskIds, err := Downloader.CreateDirectBatch(&req)
 		if err != nil {
 			WriteJson(w, model.NewErrorResult(err.Error()))
 			return

pkg/rest/model/task.go

@@ -7,8 +7,3 @@ type CreateTask struct {
 	Req *base.Request `json:"req"`
 	Opt *base.Options `json:"opt"`
 }
-
-type CreateTaskBatch struct {
-	Reqs []*base.Request `json:"reqs"`
-	Opt  *base.Options   `json:"opt"`
-}

pkg/rest/server_test.go

@@ -16,6 +16,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -140,11 +141,13 @@ func TestCreateDirectTask(t *testing.T) {
 
 func TestCreateDirectTaskBatch(t *testing.T) {
 	doTest(func() {
-		reqs := make([]*base.Request, 0)
+		reqs := make([]*base.CreateTaskBatchItem, 0)
 		for i := 0; i < 5; i++ {
-			reqs = append(reqs, createReq.Req)
+			reqs = append(reqs, &base.CreateTaskBatchItem{
+				Req: createReq.Req,
+			})
 		}
-		taskIds := httpRequestCheckOk[[]string](http.MethodPost, "/api/v1/tasks/batch", &model.CreateTaskBatch{
+		taskIds := httpRequestCheckOk[[]string](http.MethodPost, "/api/v1/tasks/batch", &base.CreateTaskBatch{
 			Reqs: reqs,
 		})
 		if len(taskIds) != len(reqs) {
@@ -153,6 +156,45 @@ func TestCreateDirectTaskBatch(t *testing.T) {
 	})
 }
 
+func TestCreateDirectTaskBatchWithOpt(t *testing.T) {
+	doTest(func() {
+		reqs := make([]*base.CreateTaskBatchItem, 0)
+		for i := 0; i < 5; i++ {
+			item := &base.CreateTaskBatchItem{
+				Req: createReq.Req,
+			}
+			if i == 0 {
+				item.Opts = &base.Options{
+					Name: "spe_opt.data",
+				}
+			}
+			reqs = append(reqs, item)
+		}
+		taskIds := httpRequestCheckOk[[]string](http.MethodPost, "/api/v1/tasks/batch", &base.CreateTaskBatch{
+			Reqs: reqs,
+			Opts: &base.Options{
+				Name: "default_opt.data",
+			},
+		})
+		if len(taskIds) != len(reqs) {
+			t.Errorf("CreateDirectTaskBatch() got = %v, want %v", len(taskIds), len(reqs))
+		}
+
+		for i, taskId := range taskIds {
+			task := httpRequestCheckOk[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil)
+			if i == 0 {
+				if !strings.Contains(task.Name(), "spe_opt") {
+					t.Errorf("CreateDirectTaskBatch() got = %v, want %v", task.Name(), "spe_opt.data")
+				}
+			} else {
+				if !strings.Contains(task.Name(), "default_opt") {
+					t.Errorf("CreateDirectTaskBatch() got = %v, want %v", task.Name(), "default_opt.data")
+				}
+			}
+		}
+	})
+}
+
 func TestCreateDirectTaskWithResoled(t *testing.T) {
 	doTest(func() {
 		var wg sync.WaitGroup
@@ -666,6 +708,7 @@ func doTest0(onStart func(cfg *model.StartConfig), handler func()) {
 			Downloader.Clear()
 		}()
 		defer func() {
+			time.Sleep(500 * time.Millisecond)
 			Downloader.Pause(nil)
 			Downloader.Delete(nil, true)
 			os.RemoveAll(cfg.StorageDir)