mirror of
https://github.com/cloudreve/Cloudreve.git
synced 2024-04-21 12:31:40 +00:00
* 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
156 lines
3.2 KiB
Go
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
|
|
}
|