|
|
package monitor
|
|
|
|
|
|
import (
|
|
|
"context"
|
|
|
"encoding/json"
|
|
|
"errors"
|
|
|
"path/filepath"
|
|
|
"strconv"
|
|
|
"time"
|
|
|
|
|
|
model "github.com/cloudreve/Cloudreve/v3/models"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/task"
|
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
|
|
)
|
|
|
|
|
|
// Monitor 离线下载状态监控
|
|
|
type Monitor struct {
|
|
|
Task *model.Download
|
|
|
Interval time.Duration
|
|
|
|
|
|
notifier <-chan mq.Message
|
|
|
node cluster.Node
|
|
|
retried int
|
|
|
}
|
|
|
|
|
|
var MAX_RETRY = 10
|
|
|
|
|
|
// NewMonitor 新建离线下载状态监控
|
|
|
func NewMonitor(task *model.Download, pool cluster.Pool, mqClient mq.MQ) {
|
|
|
monitor := &Monitor{
|
|
|
Task: task,
|
|
|
notifier: make(chan mq.Message),
|
|
|
node: pool.GetNodeByID(task.GetNodeID()),
|
|
|
}
|
|
|
|
|
|
if monitor.node != nil {
|
|
|
monitor.Interval = time.Duration(monitor.node.GetAria2Instance().GetConfig().Interval) * time.Second
|
|
|
go monitor.Loop(mqClient)
|
|
|
|
|
|
monitor.notifier = mqClient.Subscribe(monitor.Task.GID, 0)
|
|
|
} else {
|
|
|
monitor.setErrorStatus(errors.New("node not avaliable"))
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// Loop 开启监控循环
|
|
|
func (monitor *Monitor) Loop(mqClient mq.MQ) {
|
|
|
defer mqClient.Unsubscribe(monitor.Task.GID, monitor.notifier)
|
|
|
|
|
|
// 首次循环立即更新
|
|
|
interval := 50 * time.Millisecond
|
|
|
|
|
|
for {
|
|
|
select {
|
|
|
case <-monitor.notifier:
|
|
|
if monitor.Update() {
|
|
|
return
|
|
|
}
|
|
|
case <-time.After(interval):
|
|
|
interval = monitor.Interval
|
|
|
if monitor.Update() {
|
|
|
return
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// Update 更新状态，返回值表示是否退出监控
|
|
|
func (monitor *Monitor) Update() bool {
|
|
|
status, err := monitor.node.GetAria2Instance().Status(monitor.Task)
|
|
|
|
|
|
if err != nil {
|
|
|
monitor.retried++
|
|
|
util.Log().Warning("Cannot get status of download task %q: %s", monitor.Task.GID, err)
|
|
|
|
|
|
// 十次重试后认定为任务失败
|
|
|
if monitor.retried > MAX_RETRY {
|
|
|
util.Log().Warning("Cannot get status of download task %q，exceed maximum retry threshold: %s",
|
|
|
monitor.Task.GID, err)
|
|
|
monitor.setErrorStatus(err)
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
return false
|
|
|
}
|
|
|
monitor.retried = 0
|
|
|
|
|
|
// 磁力链下载需要跟随
|
|
|
if len(status.FollowedBy) > 0 {
|
|
|
util.Log().Debug("Redirected download task from %q to %q.", monitor.Task.GID, status.FollowedBy[0])
|
|
|
monitor.Task.GID = status.FollowedBy[0]
|
|
|
monitor.Task.Save()
|
|
|
return false
|
|
|
}
|
|
|
|
|
|
// 更新任务信息
|
|
|
if err := monitor.UpdateTaskInfo(status); err != nil {
|
|
|
util.Log().Warning("Failed to update status of download task %q: %s", monitor.Task.GID, err)
|
|
|
monitor.setErrorStatus(err)
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
util.Log().Debug("Remote download %q status updated to %q.", status.Gid, status.Status)
|
|
|
|
|
|
switch common.GetStatus(status) {
|
|
|
case common.Complete, common.Seeding:
|
|
|
return monitor.Complete(task.TaskPoll)
|
|
|
case common.Error:
|
|
|
return monitor.Error(status)
|
|
|
case common.Downloading, common.Ready, common.Paused:
|
|
|
return false
|
|
|
case common.Canceled:
|
|
|
monitor.Task.Status = common.Canceled
|
|
|
monitor.Task.Save()
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
default:
|
|
|
util.Log().Warning("Download task %q returns unknown status %q.", monitor.Task.GID, status.Status)
|
|
|
return true
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// UpdateTaskInfo 更新数据库中的任务信息
|
|
|
func (monitor *Monitor) UpdateTaskInfo(status rpc.StatusInfo) error {
|
|
|
originSize := monitor.Task.TotalSize
|
|
|
|
|
|
monitor.Task.GID = status.Gid
|
|
|
monitor.Task.Status = common.GetStatus(status)
|
|
|
|
|
|
// 文件大小、已下载大小
|
|
|
total, err := strconv.ParseUint(status.TotalLength, 10, 64)
|
|
|
if err != nil {
|
|
|
total = 0
|
|
|
}
|
|
|
downloaded, err := strconv.ParseUint(status.CompletedLength, 10, 64)
|
|
|
if err != nil {
|
|
|
downloaded = 0
|
|
|
}
|
|
|
monitor.Task.TotalSize = total
|
|
|
monitor.Task.DownloadedSize = downloaded
|
|
|
monitor.Task.GID = status.Gid
|
|
|
monitor.Task.Parent = status.Dir
|
|
|
|
|
|
// 下载速度
|
|
|
speed, err := strconv.Atoi(status.DownloadSpeed)
|
|
|
if err != nil {
|
|
|
speed = 0
|
|
|
}
|
|
|
|
|
|
monitor.Task.Speed = speed
|
|
|
attrs, _ := json.Marshal(status)
|
|
|
monitor.Task.Attrs = string(attrs)
|
|
|
|
|
|
if err := monitor.Task.Save(); err != nil {
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
if originSize != monitor.Task.TotalSize {
|
|
|
// 文件大小更新后，对文件限制等进行校验
|
|
|
if err := monitor.ValidateFile(); err != nil {
|
|
|
// 验证失败时取消任务
|
|
|
monitor.node.GetAria2Instance().Cancel(monitor.Task)
|
|
|
return err
|
|
|
}
|
|
|
}
|
|
|
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
// ValidateFile 上传过程中校验文件大小、文件名
|
|
|
func (monitor *Monitor) ValidateFile() error {
|
|
|
// 找到任务创建者
|
|
|
user := monitor.Task.GetOwner()
|
|
|
if user == nil {
|
|
|
return common.ErrUserNotFound
|
|
|
}
|
|
|
|
|
|
// 创建文件系统
|
|
|
fs, err := filesystem.NewFileSystem(user)
|
|
|
if err != nil {
|
|
|
return err
|
|
|
}
|
|
|
defer fs.Recycle()
|
|
|
|
|
|
// 创建上下文环境
|
|
|
file := &fsctx.FileStream{
|
|
|
Size: monitor.Task.TotalSize,
|
|
|
}
|
|
|
|
|
|
// 验证用户容量
|
|
|
if err := filesystem.HookValidateCapacity(context.Background(), fs, file); err != nil {
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
// 验证每个文件
|
|
|
for _, fileInfo := range monitor.Task.StatusInfo.Files {
|
|
|
if fileInfo.Selected == "true" {
|
|
|
// 创建上下文环境
|
|
|
fileSize, _ := strconv.ParseUint(fileInfo.Length, 10, 64)
|
|
|
file := &fsctx.FileStream{
|
|
|
Size: fileSize,
|
|
|
Name: filepath.Base(fileInfo.Path),
|
|
|
}
|
|
|
if err := filesystem.HookValidateFile(context.Background(), fs, file); err != nil {
|
|
|
return err
|
|
|
}
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
// Error 任务下载出错处理，返回是否中断监控
|
|
|
func (monitor *Monitor) Error(status rpc.StatusInfo) bool {
|
|
|
monitor.setErrorStatus(errors.New(status.ErrorMessage))
|
|
|
|
|
|
// 清理临时文件
|
|
|
monitor.RemoveTempFolder()
|
|
|
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
// RemoveTempFolder 清理下载临时目录
|
|
|
func (monitor *Monitor) RemoveTempFolder() {
|
|
|
monitor.node.GetAria2Instance().DeleteTempFile(monitor.Task)
|
|
|
}
|
|
|
|
|
|
// Complete 完成下载，返回是否中断监控
|
|
|
func (monitor *Monitor) Complete(pool task.Pool) bool {
|
|
|
// 未开始转存，提交转存任务
|
|
|
if monitor.Task.TaskID == 0 {
|
|
|
return monitor.transfer(pool)
|
|
|
}
|
|
|
|
|
|
// 做种完成
|
|
|
if common.GetStatus(monitor.Task.StatusInfo) == common.Complete {
|
|
|
transferTask, err := model.GetTasksByID(monitor.Task.TaskID)
|
|
|
if err != nil {
|
|
|
monitor.setErrorStatus(err)
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
// 转存完成，回收下载目录
|
|
|
if transferTask.Type == task.TransferTaskType && transferTask.Status >= task.Error {
|
|
|
job, err := task.NewRecycleTask(monitor.Task)
|
|
|
if err != nil {
|
|
|
monitor.setErrorStatus(err)
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
// 提交回收任务
|
|
|
pool.Submit(job)
|
|
|
|
|
|
return true
|
|
|
}
|
|
|
}
|
|
|
|
|
|
return false
|
|
|
}
|
|
|
|
|
|
func (monitor *Monitor) transfer(pool task.Pool) bool {
|
|
|
// 创建中转任务
|
|
|
file := make([]string, 0, len(monitor.Task.StatusInfo.Files))
|
|
|
sizes := make(map[string]uint64, len(monitor.Task.StatusInfo.Files))
|
|
|
for i := 0; i < len(monitor.Task.StatusInfo.Files); i++ {
|
|
|
fileInfo := monitor.Task.StatusInfo.Files[i]
|
|
|
if fileInfo.Selected == "true" {
|
|
|
file = append(file, fileInfo.Path)
|
|
|
size, _ := strconv.ParseUint(fileInfo.Length, 10, 64)
|
|
|
sizes[fileInfo.Path] = size
|
|
|
}
|
|
|
}
|
|
|
|
|
|
job, err := task.NewTransferTask(
|
|
|
monitor.Task.UserID,
|
|
|
file,
|
|
|
monitor.Task.Dst,
|
|
|
monitor.Task.Parent,
|
|
|
true,
|
|
|
monitor.node.ID(),
|
|
|
sizes,
|
|
|
)
|
|
|
if err != nil {
|
|
|
monitor.setErrorStatus(err)
|
|
|
monitor.RemoveTempFolder()
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
// 提交中转任务
|
|
|
pool.Submit(job)
|
|
|
|
|
|
// 更新任务ID
|
|
|
monitor.Task.TaskID = job.Model().ID
|
|
|
monitor.Task.Save()
|
|
|
|
|
|
return false
|
|
|
}
|
|
|
|
|
|
func (monitor *Monitor) setErrorStatus(err error) {
|
|
|
monitor.Task.Status = common.Error
|
|
|
monitor.Task.Error = err.Error()
|
|
|
monitor.Task.Save()
|
|
|
}
|
