125 lines
3.1 KiB
Go
125 lines
3.1 KiB
Go
|
package task
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"github.com/gogf/gf/v2/frame/g"
|
|||
|
|
"github.com/gogf/gf/v2/os/gctx"
|
|||
|
|
"github.com/robfig/cron/v3"
|
|||
|
|
"github.com/tiger1103/gfast/v3/api/v1/system"
|
|||
|
|
"github.com/tiger1103/gfast/v3/internal/app/system/controller"
|
|||
|
|
"github.com/tiger1103/gfast/v3/internal/app/system/dao"
|
|||
|
|
|
|||
|
|
"github.com/tiger1103/gfast/v3/internal/app/system/model"
|
|||
|
|
"golang.org/x/net/context"
|
|||
|
|
"strconv"
|
|||
|
|
"time"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
var c = cron.New(cron.WithSeconds())
|
|||
|
|
|
|||
|
|
// UavRun 启动任务
|
|||
|
|
func UavRun(ctx context.Context) {
|
|||
|
|
//1、获取到所有无人机的任务
|
|||
|
|
var taskEntity []*model.ManageTaskCronInfoRes
|
|||
|
|
err := dao.ManageTaskCron.Ctx(ctx).Scan(&taskEntity)
|
|||
|
|
if err != nil {
|
|||
|
|
g.Log().Error(ctx, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
if len(taskEntity) == 0 {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
//2、添加任务到cron中
|
|||
|
|
for _, data := range taskEntity {
|
|||
|
|
// 创建一个包装函数
|
|||
|
|
wrapFunc := func() {
|
|||
|
|
id := data.TaskId
|
|||
|
|
clientId := data.MqClientId
|
|||
|
|
TakeOffFunc(id, clientId)
|
|||
|
|
}
|
|||
|
|
oneId, err := c.AddFunc(data.Cron, wrapFunc)
|
|||
|
|
if err != nil {
|
|||
|
|
g.Log().Error(ctx, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
//3、将启动任务得到的任务ID转存到数据库中
|
|||
|
|
_, err = dao.ManageTaskCron.Ctx(ctx).Where(dao.ManageTaskCron.Columns().Id, data.Id).Update(g.Map{"cron_id": oneId})
|
|||
|
|
if err != nil {
|
|||
|
|
g.Log().Error(ctx, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
//4、启动任务
|
|||
|
|
c.Start()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// MidMissionFunc 中途追加或修改任务
|
|||
|
|
func MidMissionFunc() {
|
|||
|
|
time.Sleep(time.Second * 1) //防止sql语句还未插入进数据库
|
|||
|
|
//1、关闭所有任务
|
|||
|
|
c.Stop()
|
|||
|
|
//2、移除所有任务
|
|||
|
|
for _, entry := range c.Entries() {
|
|||
|
|
c.Remove(entry.ID)
|
|||
|
|
}
|
|||
|
|
//3、重新启动任务
|
|||
|
|
ctx := gctx.New()
|
|||
|
|
UavRun(ctx)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// OffTimerAllFunc 关闭所有定时器
|
|||
|
|
func OffTimerAllFunc(id int) {
|
|||
|
|
//1、关闭所有任务
|
|||
|
|
c.Stop()
|
|||
|
|
//2、移除所有任务
|
|||
|
|
for _, entry := range c.Entries() {
|
|||
|
|
c.Remove(entry.ID)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// OffTimerFunc 关闭指定的定时器
|
|||
|
|
func OffTimerFunc(id int) {
|
|||
|
|
c.Remove(cron.EntryID(id))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// TakeOffFunc 飞机从下发任务到执行任务一系列操作
|
|||
|
|
func TakeOffFunc(taskId int64, gatewaySn string) {
|
|||
|
|
ctx := context.Background()
|
|||
|
|
|
|||
|
|
//1、获取当前时间毫秒时间戳，然后组织数据，下发任务
|
|||
|
|
idStr := strconv.FormatInt(taskId, 10)
|
|||
|
|
req := system.FlighttaskPrepareReq{
|
|||
|
|
Id: idStr,
|
|||
|
|
GatewaySn: gatewaySn,
|
|||
|
|
TaskType: "0",
|
|||
|
|
ExecuteTime: time.Now().UnixNano() / int64(time.Millisecond),
|
|||
|
|
Remark: "定时执行飞行任务",
|
|||
|
|
TimedFlight: "ok",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
recordId, err := controller.FlighttaskPrepareFunc(ctx, &req)
|
|||
|
|
if err != nil {
|
|||
|
|
g.Log("uav").Error(ctx, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
//2、等待10秒钟(确保上一个指令执行完毕)，然后执行任务
|
|||
|
|
time.Sleep(10 * time.Second)
|
|||
|
|
taskReq := system.PerformATaskReq{
|
|||
|
|
Id: recordId,
|
|||
|
|
GatewaySn: gatewaySn,
|
|||
|
|
}
|
|||
|
|
_, err = controller.ManageTaskRecord.PerformATaskFunc(ctx, &taskReq)
|
|||
|
|
if err != nil {
|
|||
|
|
g.Log("uav").Error(ctx, err)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ValidateCronExprWithSeconds 用于验证 6 字段的 Cron 表达式
|
|||
|
|
func ValidateCronExprWithSeconds(expr string) error {
|
|||
|
|
// 创建一个支持秒级字段的解析器
|
|||
|
|
parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
|
|||
|
|
_, err := parser.Parse(expr)
|
|||
|
|
return err
|
|||
|
|
}
|