injectlib/OpenList
1
0
Fork 0
mirror of https://github.com/OpenListTeam/OpenList.git synced 2025-11-25 11:29:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: pass api_url through context (#457)

Browse Source 
* refactor: pass `api_url` through context

* 移除 LinkArgs.HttpReq

* pref(alias): 减少不必要下载代理

* 修复bug

* net: 支持1并发 分片下载
This commit is contained in:
j2rong4cn
2025-06-30 15:48:05 +08:00
committed by GitHub
parent f0236522f3
commit 103abc942e
30 changed files with 209 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/fs/archive.go
View File

@@ -49,7 +49,9 @@ func (t *ArchiveDownloadTask) GetStatus() string {
}
func (t *ArchiveDownloadTask) Run() error {
t.ReinitCtx()
if err := t.ReinitCtx(); err != nil {
return err
}
t.ClearEndTime()
t.SetStartTime(time.Now())
defer func() { t.SetEndTime(time.Now()) }()
@@ -152,7 +154,9 @@ func (t *ArchiveContentUploadTask) GetStatus() string {
}
func (t *ArchiveContentUploadTask) Run() error {
t.ReinitCtx()
if err := t.ReinitCtx(); err != nil {
return err
}
t.ClearEndTime()
t.SetStartTime(time.Now())
defer func() { t.SetEndTime(time.Now()) }()

10
internal/fs/copy.go
View File

@@ -7,15 +7,15 @@ import (
stdpath "path"
"time"
"github.com/OpenListTeam/OpenList/internal/errs"
"github.com/OpenListTeam/OpenList/internal/conf"
"github.com/OpenListTeam/OpenList/internal/driver"
"github.com/OpenListTeam/OpenList/internal/errs"
"github.com/OpenListTeam/OpenList/internal/model"
"github.com/OpenListTeam/OpenList/internal/op"
"github.com/OpenListTeam/OpenList/internal/stream"
"github.com/OpenListTeam/OpenList/internal/task"
"github.com/OpenListTeam/OpenList/pkg/utils"
"github.com/OpenListTeam/OpenList/server/common"
"github.com/pkg/errors"
"github.com/xhofe/tache"
)
@@ -40,7 +40,9 @@ func (t *CopyTask) GetStatus() string {
}
func (t *CopyTask) Run() error {
t.ReinitCtx()
if err := t.ReinitCtx(); err != nil {
return err
}
t.ClearEndTime()
t.SetStartTime(time.Now())
defer func() { t.SetEndTime(time.Now()) }()
@@ -107,6 +109,7 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool
t := &CopyTask{
TaskExtension: task.TaskExtension{
Creator: taskCreator,
ApiUrl: common.GetApiUrl(ctx),
},
srcStorage: srcStorage,
dstStorage: dstStorage,
@@ -140,6 +143,7 @@ func copyBetween2Storages(t *CopyTask, srcStorage, dstStorage driver.Driver, src
CopyTaskManager.Add(&CopyTask{
TaskExtension: task.TaskExtension{
Creator: t.GetCreator(),
ApiUrl: t.ApiUrl,
},
srcStorage: srcStorage,
dstStorage: dstStorage,

5
internal/fs/link.go
View File

@@ -7,7 +7,6 @@ import (
"github.com/OpenListTeam/OpenList/internal/model"
"github.com/OpenListTeam/OpenList/internal/op"
"github.com/OpenListTeam/OpenList/server/common"
"github.com/gin-gonic/gin"
"github.com/pkg/errors"
)
@@ -21,9 +20,7 @@ func link(ctx context.Context, path string, args model.LinkArgs) (*model.Link, m
return nil, nil, errors.WithMessage(err, "failed link")
}
if l.URL != "" && !strings.HasPrefix(l.URL, "http://") && !strings.HasPrefix(l.URL, "https://") {
if c, ok := ctx.(*gin.Context); ok {
l.URL = common.GetApiUrl(c.Request) + l.URL
}
l.URL = common.GetApiUrl(ctx) + l.URL
}
return l, obj, nil
}

152
internal/fs/move.go
View File

@@ -15,26 +15,27 @@ import (
"github.com/OpenListTeam/OpenList/internal/stream"
"github.com/OpenListTeam/OpenList/internal/task"
"github.com/OpenListTeam/OpenList/pkg/utils"
"github.com/OpenListTeam/OpenList/server/common"
"github.com/pkg/errors"
"github.com/xhofe/tache"
)
type MoveTask struct {
task.TaskExtension
Status string `json:"-"`
SrcObjPath string `json:"src_path"`
DstDirPath string `json:"dst_path"`
srcStorage driver.Driver `json:"-"`
dstStorage driver.Driver `json:"-"`
SrcStorageMp string `json:"src_storage_mp"`
DstStorageMp string `json:"dst_storage_mp"`
IsRootTask bool `json:"is_root_task"`
RootTaskID string `json:"root_task_id"`
TotalFiles int `json:"total_files"`
CompletedFiles int `json:"completed_files"`
Phase string `json:"phase"` // "copying", "verifying", "deleting", "completed"
ValidateExistence bool `json:"validate_existence"`
mu sync.RWMutex `json:"-"`
Status string `json:"-"`
SrcObjPath string `json:"src_path"`
DstDirPath string `json:"dst_path"`
srcStorage driver.Driver `json:"-"`
dstStorage driver.Driver `json:"-"`
SrcStorageMp string `json:"src_storage_mp"`
DstStorageMp string `json:"dst_storage_mp"`
IsRootTask bool `json:"is_root_task"`
RootTaskID string `json:"root_task_id"`
TotalFiles int `json:"total_files"`
CompletedFiles int `json:"completed_files"`
Phase string `json:"phase"` // "copying", "verifying", "deleting", "completed"
ValidateExistence bool `json:"validate_existence"`
mu sync.RWMutex `json:"-"`
}
type MoveProgress struct {
@@ -62,11 +63,11 @@ func (t *MoveTask) GetStatus() string {
func (t *MoveTask) GetProgress() float64 {
t.mu.RLock()
defer t.mu.RUnlock()
if t.TotalFiles == 0 {
return 0
}
switch t.Phase {
case "copying":
return float64(t.CompletedFiles*60) / float64(t.TotalFiles)
@@ -84,9 +85,9 @@ func (t *MoveTask) GetProgress() float64 {
func (t *MoveTask) GetMoveProgress() *MoveProgress {
t.mu.RLock()
defer t.mu.RUnlock()
progress := int(t.GetProgress())
return &MoveProgress{
TaskID: t.GetID(),
Phase: t.Phase,
@@ -106,16 +107,18 @@ func (t *MoveTask) updateProgress() {
}
func (t *MoveTask) Run() error {
t.ReinitCtx()
if err := t.ReinitCtx(); err != nil {
return err
}
t.ClearEndTime()
t.SetStartTime(time.Now())
defer func() {
defer func() {
t.SetEndTime(time.Now())
if t.IsRootTask {
moveProgressMap.Delete(t.GetID())
}
}()
var err error
if t.srcStorage == nil {
t.srcStorage, err = op.GetStorageByMountPath(t.SrcStorageMp)
@@ -131,13 +134,13 @@ func (t *MoveTask) Run() error {
t.mu.Lock()
t.Status = "validating source and destination"
t.mu.Unlock()
// Check if source exists
srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath)
if err != nil {
return errors.WithMessagef(err, "source file [%s] not found", stdpath.Base(t.SrcObjPath))
}
// Check if destination already exists (if validation is required)
if t.ValidateExistence {
dstFilePath := stdpath.Join(t.DstDirPath, srcObj.GetName())
@@ -155,7 +158,7 @@ func (t *MoveTask) Run() error {
t.mu.Unlock()
return t.runRootMoveTask()
}
// Use safe move logic for files
return t.safeMoveOperation(srcObj)
}
@@ -167,7 +170,7 @@ func (t *MoveTask) runRootMoveTask() error {
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] object", t.SrcObjPath)
}
if !srcObj.IsDir() {
// Source is not a directory, use regular move logic
t.mu.Lock()
@@ -175,32 +178,32 @@ func (t *MoveTask) runRootMoveTask() error {
t.mu.Unlock()
return t.safeMoveOperation(srcObj)
}
// Phase 1: Count total files and create directory structure
t.mu.Lock()
t.Phase = "preparing"
t.Status = "counting files and preparing directory structure"
t.mu.Unlock()
t.updateProgress()
totalFiles, err := t.countFilesAndCreateDirs(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
if err != nil {
return errors.WithMessage(err, "failed to prepare directory structure")
}
t.mu.Lock()
t.TotalFiles = totalFiles
t.Phase = "copying"
t.Status = "copying files"
t.mu.Unlock()
t.updateProgress()
// Phase 2: Copy all files
err = t.copyAllFiles(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
if err != nil {
return errors.WithMessage(err, "failed to copy files")
}
// Phase 3: Verify directory structure
t.mu.Lock()
t.Phase = "verifying"
@@ -208,12 +211,12 @@ func (t *MoveTask) runRootMoveTask() error {
t.CompletedFiles = 0
t.mu.Unlock()
t.updateProgress()
err = t.verifyDirectoryStructure(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
if err != nil {
return errors.WithMessage(err, "verification failed")
}
// Phase 4: Delete source files and directories
t.mu.Lock()
t.Phase = "deleting"
@@ -221,18 +224,18 @@ func (t *MoveTask) runRootMoveTask() error {
t.CompletedFiles = 0
t.mu.Unlock()
t.updateProgress()
err = t.deleteSourceRecursively(t.srcStorage, t.SrcObjPath)
if err != nil {
return errors.WithMessage(err, "failed to delete source files")
}
t.mu.Lock()
t.Phase = "completed"
t.Status = "completed"
t.mu.Unlock()
t.updateProgress()
return nil
}
@@ -257,11 +260,11 @@ func (t *MoveTask) countFilesAndCreateDirs(srcStorage, dstStorage driver.Driver,
if err != nil {
return 0, errors.WithMessagef(err, "failed get src [%s] object", srcPath)
}
if !srcObj.IsDir() {
return 1, nil
}
// Create destination directory
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
err = op.MakeDir(t.Ctx(), dstStorage, dstObjPath)
@@ -271,13 +274,13 @@ func (t *MoveTask) countFilesAndCreateDirs(srcStorage, dstStorage driver.Driver,
}
return 0, errors.WithMessagef(err, "failed to create destination directory [%s] in storage [%s]", dstObjPath, dstStorage.GetStorage().MountPath)
}
// List and count files recursively
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
if err != nil {
return 0, errors.WithMessagef(err, "failed list src [%s] objs", srcPath)
}
totalFiles := 0
for _, obj := range objs {
if utils.IsCanceled(t.Ctx()) {
@@ -290,7 +293,7 @@ func (t *MoveTask) countFilesAndCreateDirs(srcStorage, dstStorage driver.Driver,
}
totalFiles += subCount
}
return totalFiles, nil
}
@@ -300,27 +303,27 @@ func (t *MoveTask) copyAllFiles(srcStorage, dstStorage driver.Driver, srcPath, d
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] object", srcPath)
}
if !srcObj.IsDir() {
// Copy single file
err := t.copyFile(srcStorage, dstStorage, srcPath, dstPath)
if err != nil {
return err
}
t.mu.Lock()
t.CompletedFiles++
t.mu.Unlock()
t.updateProgress()
return nil
}
// Copy directory contents
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
if err != nil {
return errors.WithMessagef(err, "failed list src [%s] objs", srcPath)
}
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
for _, obj := range objs {
if utils.IsCanceled(t.Ctx()) {
@@ -332,7 +335,7 @@ func (t *MoveTask) copyAllFiles(srcStorage, dstStorage driver.Driver, srcPath, d
return err
}
}
return nil
}
@@ -342,24 +345,24 @@ func (t *MoveTask) copyFile(srcStorage, dstStorage driver.Driver, srcFilePath, d
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] file", srcFilePath)
}
link, _, err := op.Link(t.Ctx(), srcStorage, srcFilePath, model.LinkArgs{
Header: http.Header{},
})
if err != nil {
return errors.WithMessagef(err, "failed get [%s] link", srcFilePath)
}
fs := stream.FileStream{
Obj: srcFile,
Ctx: t.Ctx(),
}
ss, err := stream.NewSeekableStream(fs, link)
if err != nil {
return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath)
}
return op.Put(t.Ctx(), dstStorage, dstDirPath, ss, nil, true)
}
@@ -369,7 +372,7 @@ func (t *MoveTask) verifyDirectoryStructure(srcStorage, dstStorage driver.Driver
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] object", srcPath)
}
if !srcObj.IsDir() {
// Verify single file
dstFilePath := stdpath.Join(dstPath, srcObj.GetName())
@@ -377,27 +380,27 @@ func (t *MoveTask) verifyDirectoryStructure(srcStorage, dstStorage driver.Driver
if err != nil {
return errors.WithMessagef(err, "verification failed: destination file [%s] not found", dstFilePath)
}
t.mu.Lock()
t.CompletedFiles++
t.mu.Unlock()
t.updateProgress()
return nil
}
// Verify directory
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
_, err = op.Get(t.Ctx(), dstStorage, dstObjPath)
if err != nil {
return errors.WithMessagef(err, "verification failed: destination directory [%s] not found", dstObjPath)
}
// Verify directory contents
srcObjs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
if err != nil {
return errors.WithMessagef(err, "failed list src [%s] objs for verification", srcPath)
}
for _, obj := range srcObjs {
if utils.IsCanceled(t.Ctx()) {
return nil
@@ -408,7 +411,7 @@ func (t *MoveTask) verifyDirectoryStructure(srcStorage, dstStorage driver.Driver
return err
}
}
return nil
}
@@ -418,27 +421,27 @@ func (t *MoveTask) deleteSourceRecursively(srcStorage driver.Driver, srcPath str
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] object for deletion", srcPath)
}
if !srcObj.IsDir() {
// Delete single file
err := op.Remove(t.Ctx(), srcStorage, srcPath)
if err != nil {
return errors.WithMessagef(err, "failed to delete src [%s] file", srcPath)
}
t.mu.Lock()
t.CompletedFiles++
t.mu.Unlock()
t.updateProgress()
return nil
}
// Delete directory contents first
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
if err != nil {
return errors.WithMessagef(err, "failed list src [%s] objs for deletion", srcPath)
}
for _, obj := range objs {
if utils.IsCanceled(t.Ctx()) {
return nil
@@ -449,13 +452,13 @@ func (t *MoveTask) deleteSourceRecursively(srcStorage driver.Driver, srcPath str
return err
}
}
// Delete the directory itself
err = op.Remove(t.Ctx(), srcStorage, srcPath)
if err != nil {
return errors.WithMessagef(err, "failed to delete src [%s] directory", srcPath)
}
return nil
}
@@ -465,14 +468,14 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src
if err != nil {
return errors.WithMessagef(err, "failed get src [%s] file", srcObjPath)
}
if srcObj.IsDir() {
t.Status = "src object is dir, listing objs"
objs, err := op.List(t.Ctx(), srcStorage, srcObjPath, model.ListArgs{})
if err != nil {
return errors.WithMessagef(err, "failed list src [%s] objs", srcObjPath)
}
dstObjPath := stdpath.Join(dstDirPath, srcObj.GetName())
t.Status = "creating destination directory"
err = op.MakeDir(t.Ctx(), dstStorage, dstObjPath)
@@ -483,7 +486,7 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src
}
return errors.WithMessagef(err, "failed to create destination directory [%s] in storage [%s]", dstObjPath, dstStorage.GetStorage().MountPath)
}
for _, obj := range objs {
if utils.IsCanceled(t.Ctx()) {
return nil
@@ -492,6 +495,7 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src
MoveTaskManager.Add(&MoveTask{
TaskExtension: task.TaskExtension{
Creator: t.GetCreator(),
ApiUrl: t.ApiUrl,
},
srcStorage: srcStorage,
dstStorage: dstStorage,
@@ -515,13 +519,13 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src
}
}
func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Driver, srcFilePath, dstDirPath string) error {
tsk.Status = "copying file to destination"
copyTask := &CopyTask{
TaskExtension: task.TaskExtension{
Creator: tsk.GetCreator(),
ApiUrl: tsk.ApiUrl,
},
srcStorage: srcStorage,
dstStorage: dstStorage,
@@ -530,10 +534,8 @@ func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Drive
SrcStorageMp: srcStorage.GetStorage().MountPath,
DstStorageMp: dstStorage.GetStorage().MountPath,
}
copyTask.SetCtx(tsk.Ctx())
err := copyBetween2Storages(copyTask, srcStorage, dstStorage, srcFilePath, dstDirPath)
if err != nil {
@@ -543,21 +545,20 @@ func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Drive
}
return errors.WithMessagef(err, "failed to copy [%s] to destination storage [%s]", srcFilePath, dstStorage.GetStorage().MountPath)
}
tsk.SetProgress(50)
tsk.Status = "deleting source file"
err = op.Remove(tsk.Ctx(), srcStorage, srcFilePath)
if err != nil {
return errors.WithMessagef(err, "failed to delete src [%s] file from storage [%s] after successful copy", srcFilePath, srcStorage.GetStorage().MountPath)
}
tsk.SetProgress(100)
tsk.Status = "completed"
return nil
}
// safeMoveOperation ensures copy-then-delete sequence for safe move operations
func (t *MoveTask) safeMoveOperation(srcObj model.Obj) error {
if srcObj.IsDir() {
@@ -592,12 +593,13 @@ func _moveWithValidation(ctx context.Context, srcObjPath, dstDirPath string, val
}
taskCreator, _ := ctx.Value("user").(*model.User)
// Create task immediately without any synchronous checks to avoid blocking frontend
// All validation and type checking will be done asynchronously in the Run method
t := &MoveTask{
TaskExtension: task.TaskExtension{
Creator: taskCreator,
ApiUrl: common.GetApiUrl(ctx),
},
srcStorage: srcStorage,
dstStorage: dstStorage,
@@ -608,7 +610,7 @@ func _moveWithValidation(ctx context.Context, srcObjPath, dstDirPath string, val
ValidateExistence: validateExistence,
Phase: "initializing",
}
MoveTaskManager.Add(t)
return t, nil
}
}