@@ -3,7 +3,11 @@ package fs
33import (
44 "context"
55 "fmt"
6+ "github.com/OpenListTeam/OpenList/v4/internal/driver"
7+ "github.com/OpenListTeam/OpenList/v4/internal/stream"
8+ "github.com/OpenListTeam/OpenList/v4/pkg/utils"
69 stdpath "path"
10+ "time"
711
812 "github.com/OpenListTeam/OpenList/v4/internal/conf"
913 "github.com/OpenListTeam/OpenList/v4/internal/errs"
@@ -12,53 +16,184 @@ import (
1216 "github.com/OpenListTeam/OpenList/v4/internal/task"
1317 "github.com/OpenListTeam/OpenList/v4/internal/task/batch_task"
1418 "github.com/OpenListTeam/OpenList/v4/server/common"
19+ "github.com/OpenListTeam/tache"
1520 "github.com/pkg/errors"
1621)
1722
18- func _moveWithValidation (ctx context.Context , srcPath , dstPath string , validateExistence bool , lazyCache ... bool ) error {
23+ type MoveTask struct {
24+ task.TaskExtension
25+ Status string `json:"-"` //don't save status to save space
26+ SrcObjPath string `json:"src_path"`
27+ DstDirPath string `json:"dst_path"`
28+ srcStorage driver.Driver `json:"-"`
29+ dstStorage driver.Driver `json:"-"`
30+ SrcStorageMp string `json:"src_storage_mp"`
31+ DstStorageMp string `json:"dst_storage_mp"`
32+ }
33+
34+ func (t * MoveTask ) GetName () string {
35+ return fmt .Sprintf ("move [%s](%s) to [%s](%s)" , t .SrcStorageMp , t .SrcObjPath , t .DstStorageMp , t .DstDirPath )
36+ }
37+
38+ func (t * MoveTask ) GetStatus () string {
39+ return t .Status
40+ }
41+
42+ func (t * MoveTask ) Run () error {
43+ return task .RunWithLifecycle (t )
44+ }
45+
46+ var _ task.Lifecycle = (* MoveTask )(nil )
47+
48+ func (t * MoveTask ) BeforeRun () error {
49+ batch_task .BatchTaskRefreshAndRemoveHook .AddTask (t .GetID (), batch_task.TaskMap {
50+ batch_task .NeedRefreshPath : stdpath .Join (t .DstStorageMp , t .DstDirPath ),
51+ batch_task .MoveSrcPath : stdpath .Join (t .srcStorage .GetStorage ().MountPath , t .SrcObjPath ),
52+ batch_task .MoveDstPath : stdpath .Join (t .dstStorage .GetStorage ().MountPath , t .DstDirPath ),
53+ })
54+ return nil
55+ }
56+
57+ func (t * MoveTask ) RunCore () error {
58+ if err := t .ReinitCtx (); err != nil {
59+ return err
60+ }
61+ t .ClearEndTime ()
62+ t .SetStartTime (time .Now ())
63+ defer func () { t .SetEndTime (time .Now ()) }()
64+ var err error
65+ if t .srcStorage == nil {
66+ t .srcStorage , err = op .GetStorageByMountPath (t .SrcStorageMp )
67+ }
68+ if t .dstStorage == nil {
69+ t .dstStorage , err = op .GetStorageByMountPath (t .DstStorageMp )
70+ }
71+ if err != nil {
72+ return errors .WithMessage (err , "failed get storage" )
73+ }
74+ return moveBetween2Storages (t , t .srcStorage , t .dstStorage , t .SrcObjPath , t .DstDirPath )
75+ }
76+
77+ func (t * MoveTask ) AfterRun (err error ) error {
78+ allFinish := true
79+ // 需要先更新任务状态，再进行判断
80+ if err == nil {
81+ t .State = tache .StateSucceeded
82+ } else {
83+ t .State = tache .StateFailed
84+ }
85+ for _ , ct := range MoveTaskManager .GetAll () {
86+ if ! utils .SliceContains ([]tache.State {
87+ tache .StateSucceeded ,
88+ tache .StateFailed ,
89+ tache .StateCanceled ,
90+ }, ct .GetState ()) {
91+ allFinish = false
92+ break
93+ }
94+
95+ }
96+ batch_task .BatchTaskRefreshAndRemoveHook .RemoveTask (t .GetID (), allFinish )
97+ return err
98+ }
99+
100+ var MoveTaskManager * tache.Manager [* MoveTask ]
101+
102+ func _moveWithValidation (ctx context.Context , srcPath , dstPath string , lazyCache ... bool ) (task.TaskExtensionInfo , error ) {
19103 srcStorage , srcObjActualPath , err := op .GetStorageAndActualPath (srcPath )
20104 if err != nil {
21- return errors .WithMessage (err , "failed get src storage" )
105+ return nil , errors .WithMessage (err , "failed get src storage" )
22106 }
23107 dstStorage , dstDirActualPath , err := op .GetStorageAndActualPath (dstPath )
24108 if err != nil {
25- return errors .WithMessage (err , "failed get dst storage" )
109+ return nil , errors .WithMessage (err , "failed get dst storage" )
26110 }
27111
28112 _ , err = op .Get (ctx , srcStorage , srcObjActualPath )
29113 if err != nil {
30- return errors .WithMessagef (err , "failed get src [%s] object" , srcPath )
114+ return nil , errors .WithMessagef (err , "failed get src [%s] object" , srcPath )
31115 }
32116
33117 // Try native move first if in the same storage
34118 if srcStorage .GetStorage () == dstStorage .GetStorage () {
35119 err = op .Move (ctx , srcStorage , srcObjActualPath , dstDirActualPath , lazyCache ... )
36120 if ! errors .Is (err , errs .NotImplement ) && ! errors .Is (err , errs .NotSupport ) {
37- return err
121+ return nil , err
38122 }
39123 }
40124
41125 taskCreator , _ := ctx .Value (conf .UserKey ).(* model.User )
42- copyTask := & CopyTask {
126+ moveTask := & MoveTask {
43127 TaskExtension : task.TaskExtension {
44128 Creator : taskCreator ,
45129 ApiUrl : common .GetApiUrl (ctx ),
46130 },
47-
48131 srcStorage : srcStorage ,
49132 dstStorage : dstStorage ,
50133 SrcObjPath : srcObjActualPath ,
51134 DstDirPath : dstDirActualPath ,
52135 SrcStorageMp : srcStorage .GetStorage ().MountPath ,
53136 DstStorageMp : dstStorage .GetStorage ().MountPath ,
54137 }
138+ MoveTaskManager .Add (moveTask )
139+ return moveTask , nil
140+ }
55141
56- taskID := fmt .Sprintf ("%p" , copyTask )
57- copyTask .SetID (taskID )
58- batch_task .BatchTaskRefreshAndRemoveHook .AddTask (taskID , batch_task.TaskMap {
59- batch_task .MoveSrcPath : stdpath .Join (copyTask .SrcStorageMp , srcObjActualPath ),
60- batch_task .MoveDstPath : stdpath .Join (copyTask .DstStorageMp , dstDirActualPath ),
61- })
62- CopyTaskManager .Add (copyTask )
63- return nil
142+ func moveBetween2Storages (t * MoveTask , srcStorage , dstStorage driver.Driver , srcObjPath , dstDirPath string ) error {
143+ t .Status = "getting src object"
144+ srcObj , err := op .Get (t .Ctx (), srcStorage , srcObjPath )
145+ if err != nil {
146+ return errors .WithMessagef (err , "failed get src [%s] file" , srcObjPath )
147+ }
148+ if srcObj .IsDir () {
149+ t .Status = "src object is dir, listing objs"
150+ objs , err := op .List (t .Ctx (), srcStorage , srcObjPath , model.ListArgs {})
151+ if err != nil {
152+ return errors .WithMessagef (err , "failed list src [%s] objs" , srcObjPath )
153+ }
154+ for _ , obj := range objs {
155+ if utils .IsCanceled (t .Ctx ()) {
156+ return nil
157+ }
158+ srcObjPath := stdpath .Join (srcObjPath , obj .GetName ())
159+ dstObjPath := stdpath .Join (dstDirPath , srcObj .GetName ())
160+ MoveTaskManager .Add (& MoveTask {
161+ TaskExtension : task.TaskExtension {
162+ Creator : t .GetCreator (),
163+ ApiUrl : t .ApiUrl ,
164+ },
165+ srcStorage : srcStorage ,
166+ dstStorage : dstStorage ,
167+ SrcObjPath : srcObjPath ,
168+ DstDirPath : dstObjPath ,
169+ SrcStorageMp : srcStorage .GetStorage ().MountPath ,
170+ DstStorageMp : dstStorage .GetStorage ().MountPath ,
171+ })
172+ }
173+ t .Status = "src object is dir, added all move tasks of objs"
174+ return nil
175+ }
176+ return moveFileBetween2Storages (t , srcStorage , dstStorage , srcObjPath , dstDirPath )
177+ }
178+
179+ func moveFileBetween2Storages (tsk * MoveTask , srcStorage , dstStorage driver.Driver , srcFilePath , dstDirPath string ) error {
180+ srcFile , err := op .Get (tsk .Ctx (), srcStorage , srcFilePath )
181+ if err != nil {
182+ return errors .WithMessagef (err , "failed get src [%s] file" , srcFilePath )
183+ }
184+ tsk .SetTotalBytes (srcFile .GetSize ())
185+ link , _ , err := op .Link (tsk .Ctx (), srcStorage , srcFilePath , model.LinkArgs {})
186+ if err != nil {
187+ return errors .WithMessagef (err , "failed get [%s] link" , srcFilePath )
188+ }
189+ // any link provided is seekable
190+ ss , err := stream .NewSeekableStream (& stream.FileStream {
191+ Obj : srcFile ,
192+ Ctx : tsk .Ctx (),
193+ }, link )
194+ if err != nil {
195+ _ = link .Close ()
196+ return errors .WithMessagef (err , "failed get [%s] stream" , srcFilePath )
197+ }
198+ return op .Put (tsk .Ctx (), dstStorage , dstDirPath , ss , tsk .SetProgress , true )
64199}
0 commit comments