sparrow/services/onekey-service/internal/scheduler.go

package internal

import (
	"encoding/json"
	"github.com/gogf/gf/container/gmap"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/v2/os/gcron"
	"sparrow/pkg/rule"
	"sparrow/pkg/server"
)

// TaskSchedule task schedule 任务调度
type TaskSchedule struct {
	tasks *gmap.HashMap // 保存任务名称与任务实体的映射
}

func (t *TaskSchedule) AddMessageHandle(msg *rule.TaskLifecycleMessage) error {
	return nil
}

func NewTaskSchedule() *TaskSchedule {
	return &TaskSchedule{
		tasks: gmap.NewHashMap(true),
	}
}

func (t *TaskSchedule) DoTask(msg []byte) error {
	var task rule.TimerTaskMessage
	err := json.Unmarshal([]byte(msg), &task)
	if err != nil {
		return err
	}

	newJson := gjson.New(task.Actions)
	server.Log.Debugf("do task:actions:%s", newJson.MustToJsonString())
	// 创建任务
	err = NewTaskExecutor(task.Actions).Do()
	if err != nil {
		return err
	}

	return nil
}

func (t *TaskSchedule) RemoveMessageHandle(msg *rule.TaskLifecycleMessage) error {

	gcron.Remove(msg.TaskId)
	_ = t.tasks.Remove(msg.TaskId)

	server.Log.Debugf("RemoveMessageHandle :%s", msg.TaskId)
	return nil
}

func (t *TaskSchedule) UpdateMessageHandle(msg *rule.TaskLifecycleMessage) error {

	server.Log.Debugf("UpdateMessageHandle :%s", msg.TaskId)
	return nil
}

func (t *TaskSchedule) SnapMessageHandle(msg *rule.TaskLifecycleMessage) error {
	gcron.Stop(msg.TaskId)

	server.Log.Debugf("SnapMessageHandle :%s", msg.TaskId)
	return nil
}

func (t *TaskSchedule) StartMessageHandle(msg *rule.TaskLifecycleMessage) error {
	gcron.Start(msg.TaskId)
	server.Log.Debugf("StartMessageHandle :%s", msg.TaskId)
	return nil
}

func (t *TaskSchedule) StopMessageHandle(msg *rule.TaskLifecycleMessage) error {
	gcron.Stop(msg.TaskId)
	server.Log.Debugf("StopMessageHandle :%s", msg.TaskId)
	return nil
}