ProjectHentai/alist
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(aria2): export task manager

Browse Source
This commit is contained in:
Noah Hsu 2022-06-23 21:24:23 +08:00
parent 92983aa185
commit 40b7ecc845
5 changed files with 17 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/aria2/add.go
View File

@ -43,7 +43,7 @@ func AddURI(ctx context.Context, uri string, dstDirPath string) error {
if err != nil { if err != nil {
return errors.Wrapf(err, "failed to add uri %s", uri) return errors.Wrapf(err, "failed to add uri %s", uri)
} }
downTaskManager.Submit(task.WithCancelCtx(&task.Task[string]{ DownTaskManager.Submit(task.WithCancelCtx(&task.Task[string]{
ID: gid, ID: gid,
Name: fmt.Sprintf("download %s to [%s](%s)", uri, account.GetAccount().VirtualPath, dstDirActualPath), Name: fmt.Sprintf("download %s to [%s](%s)", uri, account.GetAccount().VirtualPath, dstDirActualPath),
Func: func(tsk *task.Task[string]) error { Func: func(tsk *task.Task[string]) error {

2
internal/aria2/aria2.go
View File

@ -9,7 +9,7 @@ import (
"time" "time"
) )
var downTaskManager = task.NewTaskManager[string](3) var DownTaskManager = task.NewTaskManager[string](3)
var notify = NewNotify() var notify = NewNotify()
var client rpc.Client var client rpc.Client

6
internal/aria2/aria2_test.go
View File

@ -54,7 +54,7 @@ func TestDown(t *testing.T) {
if err != nil { if err != nil {
t.Errorf("failed to add uri: %+v", err) t.Errorf("failed to add uri: %+v", err)
} }
tasks := downTaskManager.GetAll() tasks := DownTaskManager.GetAll()
if len(tasks) != 1 { if len(tasks) != 1 {
t.Errorf("failed to get tasks: %+v", tasks) t.Errorf("failed to get tasks: %+v", tasks)
} }
@ -70,10 +70,10 @@ func TestDown(t *testing.T) {
time.Sleep(time.Second) time.Sleep(time.Second)
} }
for { for {
if len(transferTaskManager.GetAll()) == 0 { if len(TransferTaskManager.GetAll()) == 0 {
continue continue
} }
tsk := transferTaskManager.GetAll()[0] tsk := TransferTaskManager.GetAll()[0]
t.Logf("task: %+v", tsk) t.Logf("task: %+v", tsk)
if tsk.GetState() == task.Succeeded { if tsk.GetState() == task.Succeeded {
break break

13
internal/aria2/manage.go
View File

@ -1,7 +1,14 @@
package aria2 package aria2
import "context" import (
"context"
func ListFinished(ctx context.Context) { "github.com/alist-org/alist/v3/pkg/task"
)
func ListFinished(ctx context.Context) []*task.Task[string] {
return DownTaskManager.GetByStates(task.Succeeded, task.CANCELED, task.ERRORED)
}
func ListUndone(ctx context.Context) []*task.Task[string] {
return DownTaskManager.GetByStates(task.PENDING, task.RUNNING, task.CANCELING)
} }

4
internal/aria2/monitor.go
View File

@ -92,7 +92,7 @@ func (m *Monitor) Update() (bool, error) {
} }
} }
var transferTaskManager = task.NewTaskManager[uint64](3, func(k *uint64) { var TransferTaskManager = task.NewTaskManager[uint64](3, func(k *uint64) {
atomic.AddUint64(k, 1) atomic.AddUint64(k, 1)
}) })
@ -119,7 +119,7 @@ func (m *Monitor) Complete() error {
} }
}() }()
for _, file := range files { for _, file := range files {
transferTaskManager.Submit(task.WithCancelCtx[uint64](&task.Task[uint64]{ TransferTaskManager.Submit(task.WithCancelCtx[uint64](&task.Task[uint64]{
Name: fmt.Sprintf("transfer %s to [%s](%s)", file.Path, account.GetAccount().VirtualPath, dstDirActualPath), Name: fmt.Sprintf("transfer %s to [%s](%s)", file.Path, account.GetAccount().VirtualPath, dstDirActualPath),
Func: func(tsk *task.Task[uint64]) error { Func: func(tsk *task.Task[uint64]) error {
defer wg.Done() defer wg.Done()