Files
Cloudreve/pkg/task/compress.go
T
AaronLiuandGitHub 056de22edb Feat: aria2 download and transfer in slave node (#1040)
* Feat: retrieve nodes from data table

* Feat: master node ping slave node in REST API

* Feat: master send scheduled ping request

* Feat: inactive nodes recover loop

* Modify: remove database operations from aria2 RPC caller implementation

* Feat: init aria2 client in master node

* Feat: Round Robin load balancer

* Feat: create and monitor aria2 task in master node

* Feat: salve receive and handle heartbeat

* Fix: Node ID will be 0 in download record generated in older version

* Feat: sign request headers with all `X-` prefix

* Feat: API call to slave node will carry meta data in headers

* Feat: call slave aria2 rpc method from master

* Feat: get slave aria2 task status
Feat: encode slave response data using gob

* Feat: aria2 callback to master node / cancel or select task to slave node

* Fix: use dummy aria2 client when caller initialize failed in master node

* Feat: slave aria2 status event callback / salve RPC auth

* Feat: prototype for slave driven filesystem

* Feat: retry for init aria2 client in master node

* Feat: init request client with global options

* Feat: slave receive async task from master

* Fix: competition write in request header

* Refactor: dependency initialize order

* Feat: generic message queue implementation

* Feat: message queue implementation

* Feat: master waiting slave transfer result

* Feat: slave transfer file in stateless policy

* Feat: slave transfer file in slave policy

* Feat: slave transfer file in local policy

* Feat: slave transfer file in OneDrive policy

* Fix: failed to initialize update checker http client

* Feat: list slave nodes for dashboard

* Feat: test aria2 rpc connection in slave

* Feat: add and save node

* Feat: add and delete node in node pool

* Fix: temp file cannot be removed when aria2 task fails

* Fix: delete node in admin panel

* Feat: edit node and get node info

* Modify: delete unused settings
2021-10-31 09:41:56 +08:00

156 lines
3.2 KiB
Go

package task
import (
"context"
"encoding/json"
"os"
model "github.com/cloudreve/Cloudreve/v3/models"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
"github.com/cloudreve/Cloudreve/v3/pkg/util"
)
// CompressTask 文件压缩任务
type CompressTask struct {
User *model.User
TaskModel *model.Task
TaskProps CompressProps
Err *JobError
zipPath string
}
// CompressProps 压缩任务属性
type CompressProps struct {
Dirs []uint `json:"dirs"`
Files []uint `json:"files"`
Dst string `json:"dst"`
}
// Props 获取任务属性
func (job *CompressTask) Props() string {
res, _ := json.Marshal(job.TaskProps)
return string(res)
}
// Type 获取任务状态
func (job *CompressTask) Type() int {
return CompressTaskType
}
// Creator 获取创建者ID
func (job *CompressTask) Creator() uint {
return job.User.ID
}
// Model 获取任务的数据库模型
func (job *CompressTask) Model() *model.Task {
return job.TaskModel
}
// SetStatus 设定状态
func (job *CompressTask) SetStatus(status int) {
job.TaskModel.SetStatus(status)
}
// SetError 设定任务失败信息
func (job *CompressTask) SetError(err *JobError) {
job.Err = err
res, _ := json.Marshal(job.Err)
job.TaskModel.SetError(string(res))
// 删除压缩文件
job.removeZipFile()
}
func (job *CompressTask) removeZipFile() {
if job.zipPath != "" {
if err := os.Remove(job.zipPath); err != nil {
util.Log().Warning("无法删除临时压缩文件 %s , %s", job.zipPath, err)
}
}
}
// SetErrorMsg 设定任务失败信息
func (job *CompressTask) SetErrorMsg(msg string) {
job.SetError(&JobError{Msg: msg})
}
// GetError 返回任务失败信息
func (job *CompressTask) GetError() *JobError {
return job.Err
}
// Do 开始执行任务
func (job *CompressTask) Do() {
// 创建文件系统
fs, err := filesystem.NewFileSystem(job.User)
if err != nil {
job.SetErrorMsg(err.Error())
return
}
util.Log().Debug("开始压缩文件")
job.TaskModel.SetProgress(CompressingProgress)
// 开始压缩
ctx := context.Background()
zipFile, err := fs.Compress(ctx, job.TaskProps.Dirs, job.TaskProps.Files, false)
if err != nil {
job.SetErrorMsg(err.Error())
return
}
job.zipPath = zipFile
util.Log().Debug("压缩文件存放至%s,开始上传", zipFile)
job.TaskModel.SetProgress(TransferringProgress)
// 上传文件
err = fs.UploadFromPath(ctx, zipFile, job.TaskProps.Dst, true)
if err != nil {
job.SetErrorMsg(err.Error())
return
}
job.removeZipFile()
}
// NewCompressTask 新建压缩任务
func NewCompressTask(user *model.User, dst string, dirs, files []uint) (Job, error) {
newTask := &CompressTask{
User: user,
TaskProps: CompressProps{
Dirs: dirs,
Files: files,
Dst: dst,
},
}
record, err := Record(newTask)
if err != nil {
return nil, err
}
newTask.TaskModel = record
return newTask, nil
}
// NewCompressTaskFromModel 从数据库记录中恢复压缩任务
func NewCompressTaskFromModel(task *model.Task) (Job, error) {
user, err := model.GetActiveUserByID(task.UserID)
if err != nil {
return nil, err
}
newTask := &CompressTask{
User: &user,
TaskModel: task,
}
err = json.Unmarshal([]byte(task.Props), &newTask.TaskProps)
if err != nil {
return nil, err
}
return newTask, nil
}