JCC-CSScheduler/executor/internal/task/task.go

package task

import (
	"fmt"
	"reflect"

	"gitlink.org.cn/cloudream/common/pkgs/task"
	myreflect "gitlink.org.cn/cloudream/common/utils/reflect"
	exectsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor/task"
	reporter "gitlink.org.cn/cloudream/scheduler/executor/internal/reporter"
)

type TaskContext struct {
	reporter *reporter.Reporter
}

// 需要在Task结束后主动调用，completing函数将在Manager加锁期间被调用，
// 因此适合进行执行结果的设置
type CompleteFn = task.CompleteFn

type Manager struct {
	task.Manager[TaskContext]
}

type TaskBody = task.TaskBody[TaskContext]

type Task = task.Task[TaskContext]

type CompleteOption = task.CompleteOption

func NewManager(reporter *reporter.Reporter) Manager {
	return Manager{
		Manager: task.NewManager(TaskContext{
			reporter: reporter,
		}),
	}
}

func (m *Manager) StartByInfo(info exectsk.TaskInfo) (*Task, error) {
	infoType := myreflect.TypeOfValue(info)

	ctor, ok := taskFromInfoCtors[infoType]
	if !ok {
		return nil, fmt.Errorf("unknow info type")
	}

	return m.StartNew(ctor(info)), nil
}

var taskFromInfoCtors map[reflect.Type]func(exectsk.TaskInfo) TaskBody = make(map[reflect.Type]func(exectsk.TaskInfo) task.TaskBody[TaskContext])

func Register[TInfo exectsk.TaskInfo, TTaskBody TaskBody](ctor func(info TInfo) TTaskBody) {
	taskFromInfoCtors[myreflect.TypeOf[TInfo]()] = func(info exectsk.TaskInfo) TaskBody {
		return ctor(info.(TInfo))
	}
}