XYenon
3 years ago
parent
96daed26b4
commit
8f955f130f
@ -1 +1 @@
|
||||
Subproject commit a1028e7e0ae96be4bb67d8c117cf39e07c207473
|
||||
Subproject commit 02d93206cc5b943c34b5f5ac86c23dd96f5ef603
|
||||
@ -0,0 +1,155 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver/shadow/slaveinmaster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
)
|
||||
|
||||
// RecycleTask 文件回收任务
|
||||
type RecycleTask struct {
|
||||
User *model.User
|
||||
TaskModel *model.Task
|
||||
TaskProps RecycleProps
|
||||
Err *JobError
|
||||
}
|
||||
|
||||
// RecycleProps 回收任务属性
|
||||
type RecycleProps struct {
|
||||
// 回收目录
|
||||
Path string `json:"path"`
|
||||
// 负责处理回收任务的节点ID
|
||||
NodeID uint `json:"node_id"`
|
||||
}
|
||||
|
||||
// Props 获取任务属性
|
||||
func (job *RecycleTask) Props() string {
|
||||
res, _ := json.Marshal(job.TaskProps)
|
||||
return string(res)
|
||||
}
|
||||
|
||||
// Type 获取任务状态
|
||||
func (job *RecycleTask) Type() int {
|
||||
return RecycleTaskType
|
||||
}
|
||||
|
||||
// Creator 获取创建者ID
|
||||
func (job *RecycleTask) Creator() uint {
|
||||
return job.User.ID
|
||||
}
|
||||
|
||||
// Model 获取任务的数据库模型
|
||||
func (job *RecycleTask) Model() *model.Task {
|
||||
return job.TaskModel
|
||||
}
|
||||
|
||||
// SetStatus 设定状态
|
||||
func (job *RecycleTask) SetStatus(status int) {
|
||||
job.TaskModel.SetStatus(status)
|
||||
}
|
||||
|
||||
// SetError 设定任务失败信息
|
||||
func (job *RecycleTask) SetError(err *JobError) {
|
||||
job.Err = err
|
||||
res, _ := json.Marshal(job.Err)
|
||||
job.TaskModel.SetError(string(res))
|
||||
|
||||
}
|
||||
|
||||
// SetErrorMsg 设定任务失败信息
|
||||
func (job *RecycleTask) SetErrorMsg(msg string, err error) {
|
||||
jobErr := &JobError{Msg: msg}
|
||||
if err != nil {
|
||||
jobErr.Error = err.Error()
|
||||
}
|
||||
job.SetError(jobErr)
|
||||
}
|
||||
|
||||
// GetError 返回任务失败信息
|
||||
func (job *RecycleTask) GetError() *JobError {
|
||||
return job.Err
|
||||
}
|
||||
|
||||
// Do 开始执行任务
|
||||
func (job *RecycleTask) Do() {
|
||||
if job.TaskProps.NodeID == 1 {
|
||||
err := os.RemoveAll(job.TaskProps.Path)
|
||||
if err != nil {
|
||||
util.Log().Warning("无法删除中转临时目录[%s], %s", job.TaskProps.Path, err)
|
||||
job.SetErrorMsg("文件回收失败", err)
|
||||
}
|
||||
} else {
|
||||
// 指定为从机回收
|
||||
|
||||
// 创建文件系统
|
||||
fs, err := filesystem.NewFileSystem(job.User)
|
||||
if err != nil {
|
||||
job.SetErrorMsg(err.Error(), nil)
|
||||
return
|
||||
}
|
||||
|
||||
// 获取从机节点
|
||||
node := cluster.Default.GetNodeByID(job.TaskProps.NodeID)
|
||||
if node == nil {
|
||||
job.SetErrorMsg("从机节点不可用", nil)
|
||||
}
|
||||
|
||||
// 切换为从机节点处理回收
|
||||
fs.SwitchToSlaveHandler(node)
|
||||
handler := fs.Handler.(*slaveinmaster.Driver)
|
||||
err = handler.Recycle(context.Background(), job.TaskProps.Path)
|
||||
if err != nil {
|
||||
util.Log().Warning("无法删除中转临时目录[%s], %s", job.TaskProps.Path, err)
|
||||
job.SetErrorMsg("文件回收失败", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NewRecycleTask 新建回收任务
|
||||
func NewRecycleTask(user uint, path string, node uint) (Job, error) {
|
||||
creator, err := model.GetActiveUserByID(user)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
newTask := &RecycleTask{
|
||||
User: &creator,
|
||||
TaskProps: RecycleProps{
|
||||
Path: path,
|
||||
NodeID: node,
|
||||
},
|
||||
}
|
||||
|
||||
record, err := Record(newTask)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
newTask.TaskModel = record
|
||||
|
||||
return newTask, nil
|
||||
}
|
||||
|
||||
// NewRecycleTaskFromModel 从数据库记录中恢复回收任务
|
||||
func NewRecycleTaskFromModel(task *model.Task) (Job, error) {
|
||||
user, err := model.GetActiveUserByID(task.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
newTask := &RecycleTask{
|
||||
User: &user,
|
||||
TaskModel: task,
|
||||
}
|
||||
|
||||
err = json.Unmarshal([]byte(task.Props), &newTask.TaskProps)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return newTask, nil
|
||||
}
|
||||
@ -0,0 +1,131 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/jinzhu/gorm"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestRecycleTask_Props(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
task := &RecycleTask{
|
||||
User: &model.User{},
|
||||
}
|
||||
asserts.NotEmpty(task.Props())
|
||||
asserts.Equal(RecycleTaskType, task.Type())
|
||||
asserts.EqualValues(0, task.Creator())
|
||||
asserts.Nil(task.Model())
|
||||
}
|
||||
|
||||
func TestRecycleTask_SetStatus(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
task := &RecycleTask{
|
||||
User: &model.User{},
|
||||
TaskModel: &model.Task{
|
||||
Model: gorm.Model{ID: 1},
|
||||
},
|
||||
}
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec("UPDATE(.+)").WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectCommit()
|
||||
task.SetStatus(3)
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
}
|
||||
|
||||
func TestRecycleTask_SetError(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
task := &RecycleTask{
|
||||
User: &model.User{},
|
||||
TaskModel: &model.Task{
|
||||
Model: gorm.Model{ID: 1},
|
||||
},
|
||||
}
|
||||
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec("UPDATE(.+)").WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectCommit()
|
||||
|
||||
task.SetErrorMsg("error", nil)
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.Equal("error", task.GetError().Msg)
|
||||
}
|
||||
|
||||
func TestRecycleTask_Do(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
task := &RecycleTask{
|
||||
TaskModel: &model.Task{
|
||||
Model: gorm.Model{ID: 1},
|
||||
},
|
||||
}
|
||||
|
||||
// 目录不存在
|
||||
{
|
||||
task.TaskProps.Path = "test/not_exist"
|
||||
task.User = &model.User{
|
||||
Policy: model.Policy{
|
||||
Type: "unknown",
|
||||
},
|
||||
}
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec("UPDATE(.+)").WillReturnResult(sqlmock.NewResult(1,
|
||||
1))
|
||||
mock.ExpectCommit()
|
||||
task.Do()
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.NotEmpty(task.GetError().Msg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewRecycleTask(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
|
||||
// 成功
|
||||
{
|
||||
mock.ExpectQuery("SELECT(.+)").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec("INSERT(.+)").WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectCommit()
|
||||
job, err := NewRecycleTask(1, "/", 0)
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.NotNil(job)
|
||||
asserts.NoError(err)
|
||||
}
|
||||
|
||||
// 失败
|
||||
{
|
||||
mock.ExpectQuery("SELECT(.+)").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec("INSERT(.+)").WillReturnError(errors.New("error"))
|
||||
mock.ExpectRollback()
|
||||
job, err := NewRecycleTask(1, "test/not_exist", 0)
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.Nil(job)
|
||||
asserts.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewRecycleTaskFromModel(t *testing.T) {
|
||||
asserts := assert.New(t)
|
||||
|
||||
// 成功
|
||||
{
|
||||
mock.ExpectQuery("SELECT(.+)").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
|
||||
job, err := NewRecycleTaskFromModel(&model.Task{Props: "{}"})
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.NoError(err)
|
||||
asserts.NotNil(job)
|
||||
}
|
||||
|
||||
// JSON解析失败
|
||||
{
|
||||
mock.ExpectQuery("SELECT(.+)").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
|
||||
job, err := NewRecycleTaskFromModel(&model.Task{Props: "?"})
|
||||
asserts.NoError(mock.ExpectationsWereMet())
|
||||
asserts.Error(err)
|
||||
asserts.Nil(job)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,95 @@
|
||||
package slavetask
|
||||
|
||||
import (
|
||||
"os"
|
||||
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/task"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
)
|
||||
|
||||
// RecycleTask 文件回收任务
|
||||
type RecycleTask struct {
|
||||
Err *task.JobError
|
||||
Req *serializer.SlaveRecycleReq
|
||||
MasterID string
|
||||
}
|
||||
|
||||
// Props 获取任务属性
|
||||
func (job *RecycleTask) Props() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Type 获取任务类型
|
||||
func (job *RecycleTask) Type() int {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Creator 获取创建者ID
|
||||
func (job *RecycleTask) Creator() uint {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Model 获取任务的数据库模型
|
||||
func (job *RecycleTask) Model() *model.Task {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetStatus 设定状态
|
||||
func (job *RecycleTask) SetStatus(status int) {
|
||||
}
|
||||
|
||||
// SetError 设定任务失败信息
|
||||
func (job *RecycleTask) SetError(err *task.JobError) {
|
||||
job.Err = err
|
||||
}
|
||||
|
||||
// SetErrorMsg 设定任务失败信息
|
||||
func (job *RecycleTask) SetErrorMsg(msg string, err error) {
|
||||
jobErr := &task.JobError{Msg: msg}
|
||||
if err != nil {
|
||||
jobErr.Error = err.Error()
|
||||
}
|
||||
|
||||
job.SetError(jobErr)
|
||||
|
||||
notifyMsg := mq.Message{
|
||||
TriggeredBy: job.MasterID,
|
||||
Event: serializer.SlaveRecycleFailed,
|
||||
Content: serializer.SlaveRecycleResult{
|
||||
Error: err.Error(),
|
||||
},
|
||||
}
|
||||
|
||||
if err = cluster.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), notifyMsg); err != nil {
|
||||
util.Log().Warning("无法发送回收失败通知到从机, %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
// GetError 返回任务失败信息
|
||||
func (job *RecycleTask) GetError() *task.JobError {
|
||||
return job.Err
|
||||
}
|
||||
|
||||
// Do 开始执行任务
|
||||
func (job *RecycleTask) Do() {
|
||||
err := os.RemoveAll(job.Req.Path)
|
||||
if err != nil {
|
||||
util.Log().Warning("无法删除中转临时文件[%s], %s", job.Req.Path, err)
|
||||
job.SetErrorMsg("文件回收失败", err)
|
||||
return
|
||||
}
|
||||
|
||||
msg := mq.Message{
|
||||
TriggeredBy: job.MasterID,
|
||||
Event: serializer.SlaveRecycleSuccess,
|
||||
Content: serializer.SlaveRecycleResult{},
|
||||
}
|
||||
|
||||
if err = cluster.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), msg); err != nil {
|
||||
util.Log().Warning("无法发送回收成功通知到从机, %s", err)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue