149 lines
3.7 KiB
Go
149 lines
3.7 KiB
Go
package dtask
|
|
|
|
import (
|
|
"encoding/gob"
|
|
"errors"
|
|
"reflect"
|
|
"time"
|
|
|
|
"git.rosy.net.cn/baseapi/utils"
|
|
"git.rosy.net.cn/jx-callback/business/jxutils"
|
|
"git.rosy.net.cn/jx-callback/business/model"
|
|
"git.rosy.net.cn/jx-callback/business/model/dao"
|
|
"git.rosy.net.cn/jx-callback/business/partner"
|
|
)
|
|
|
|
var (
|
|
CurMan *DurableTaskMan
|
|
)
|
|
|
|
type DurableTaskMan struct {
|
|
objCreator func(objHint string) interface{}
|
|
tasks map[string]*DurableTask
|
|
}
|
|
|
|
type DurableTask struct {
|
|
cmdChan chan string
|
|
data *model.DurableTask
|
|
itemCount int
|
|
items []*model.DurableTaskItem
|
|
}
|
|
|
|
func Init(objCreator func(objHint string) interface{}, interfaceTypes ...interface{}) {
|
|
if CurMan == nil {
|
|
if objCreator == nil {
|
|
objCreator = defObjCreator
|
|
}
|
|
CurMan = &DurableTaskMan{
|
|
tasks: make(map[string]*DurableTask),
|
|
objCreator: objCreator,
|
|
}
|
|
for _, v := range interfaceTypes {
|
|
gob.Register(v)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *DurableTaskMan) LoadPendingTask() (err error) {
|
|
db := dao.GetDB()
|
|
tasks := make([]*model.DurableTask, 0)
|
|
if err = dao.GetEntitiesByKV(db, &tasks, utils.Params2Map(model.FieldStatus, 0), true); err == nil {
|
|
for _, task := range tasks {
|
|
dTask := &DurableTask{
|
|
data: task,
|
|
}
|
|
m.tasks[task.TaskID] = dTask
|
|
return dao.GetEntitiesByKV(db, &dTask.items, utils.Params2Map(model.FieldStatus, 0), true)
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (m *DurableTaskMan) Start() error {
|
|
for _, task := range m.tasks {
|
|
m.StartTask(task.data.TaskID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *DurableTaskMan) AddTask(desc, createdBy string) (taskID string, err error) {
|
|
task := &DurableTask{}
|
|
task.data = &model.DurableTask{
|
|
TaskID: utils.GetUUID(),
|
|
Description: desc,
|
|
CreatedBy: createdBy,
|
|
Status: 0,
|
|
FinishedAt: utils.DefaultTimeValue,
|
|
}
|
|
if err = dao.CreateEntity(nil, task.data); err == nil {
|
|
m.tasks[task.data.TaskID] = task
|
|
return task.data.TaskID, nil
|
|
}
|
|
return "", err
|
|
}
|
|
|
|
func (m *DurableTaskMan) AddItem(taskID, objHint string, funcName string, params ...interface{}) (err error) {
|
|
d := m.tasks[taskID]
|
|
d.data.TotalItem++
|
|
item := &model.DurableTaskItem{
|
|
ObjHint: objHint,
|
|
FuncName: funcName,
|
|
TaskID: d.data.TaskID,
|
|
TaskIndex: d.data.TotalItem,
|
|
FinishedAt: utils.DefaultTimeValue,
|
|
}
|
|
if item.Params, err = jxutils.SerializeData(params); err == nil {
|
|
db := dao.GetDB()
|
|
if err = dao.CreateEntity(db, item); err == nil {
|
|
_, err = dao.UpdateEntity(db, d.data, "TotalItem")
|
|
d.items = append(d.items, item)
|
|
}
|
|
}
|
|
d.data.TotalItem--
|
|
return err
|
|
}
|
|
|
|
func (m *DurableTaskMan) StartTask(taskID string) error {
|
|
d := m.tasks[taskID]
|
|
if d.cmdChan == nil {
|
|
d.cmdChan = make(chan string)
|
|
utils.CallFuncAsync(func() {
|
|
failedItemCount := 0
|
|
for _, taskItem := range d.items {
|
|
if taskItem.Status == 0 {
|
|
obj := m.objCreator(taskItem.ObjHint)
|
|
objValue := reflect.ValueOf(obj)
|
|
func2Call := objValue.MethodByName(taskItem.FuncName)
|
|
params := []interface{}{}
|
|
jxutils.DeSerializeData(taskItem.Params, ¶ms)
|
|
valueParams := make([]reflect.Value, len(params))
|
|
|
|
for k, v := range params {
|
|
valueParams[k] = reflect.ValueOf(v)
|
|
}
|
|
if func2Call.Call(valueParams)[0].IsNil() {
|
|
taskItem.Status = 1
|
|
taskItem.FinishedAt = time.Now()
|
|
taskItem.UpdatedAt = taskItem.FinishedAt
|
|
dao.UpdateEntity(nil, taskItem, "Status", "FinishedAt", "UpdatedAt")
|
|
} else {
|
|
failedItemCount++
|
|
}
|
|
}
|
|
}
|
|
if failedItemCount == 0 {
|
|
d.data.Status = 1
|
|
d.data.FinishedAt = time.Now()
|
|
d.data.UpdatedAt = d.data.FinishedAt
|
|
dao.UpdateEntity(nil, d.data, "Status", "FinishedAt", "UpdatedAt")
|
|
}
|
|
})
|
|
return nil
|
|
}
|
|
return errors.New("任务已经启动")
|
|
}
|
|
|
|
func defObjCreator(objHint string) interface{} {
|
|
return partner.GetPurchasePlatformFromVendorID(0)
|
|
}
|