eta_mini_ht_api/task/base/eta_task.go

package base

import (
	"context"
	"encoding/json"
	taskService "eta/eta_mini_ht_api/domian/task"
	"time"
)

const (
	DEV  = "dev"
	TEST = "test"
	PROD = "prod"
)

type TaskType string
type ETATask struct {
	RunMode string
	Name    TaskType
	TaskFn  Task
	Cron    string
}
type TaskDetail struct {
	TaskId    int
	StartTime string
	EndTime   string
	Content   string
}

type Task interface {
	Execute(taskDetail *TaskDetail) error
}

func NewTask(name TaskType, cron string, taskFunc Task, mode string) ETATask {
	return ETATask{
		Name:    name,
		TaskFn:  taskFunc,
		Cron:    cron,
		RunMode: mode,
	}
}

var tasks = make(map[TaskType]*ETATask)

func GetTasks() map[TaskType]*ETATask {
	return tasks
}
func RegisterTask(etaTask *ETATask) {
	if etaTask == nil {
		panic("不能注册空任务")
	}
	if _, ok := tasks[etaTask.Name]; ok {
		panic("请勿重复创建任务:" + etaTask.Name)
	}
	tasks[etaTask.Name] = etaTask
}

func (eta *ETATask) Run(_ context.Context) (err error) {
	details := TaskDetail{
		StartTime: time.Now().Format(time.DateTime),
	}
	runTask := taskService.TaskDTO{
		TaskType: string(eta.Name),
	}
	//启动任务
	taskId, err := taskService.InitAndStartTask(runTask)
	details.TaskId = taskId
	err = eta.TaskFn.Execute(&details)
	details.EndTime = time.Now().Format(time.DateTime)
	detailStr, err := json.Marshal(details)
	//任务状态
	if err != nil {
		taskService.UpdateAndFailedTask(taskId, string(detailStr))
		return
	}
	taskService.UpdateAndSuccessTask(taskId, string(detailStr))
	return
}