86 lines
2.2 KiB
Go
86 lines
2.2 KiB
Go
package tasksch
|
|
|
|
import (
|
|
"time"
|
|
|
|
"git.rosy.net.cn/baseapi/utils"
|
|
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
|
|
"git.rosy.net.cn/jx-callback/globals"
|
|
)
|
|
|
|
const (
|
|
StepBegin = "begin"
|
|
StepEnd = "End"
|
|
)
|
|
|
|
type SeqWorkFunc func(task *SeqTask, step int, params ...interface{}) (result interface{}, err error)
|
|
|
|
type SeqTask struct {
|
|
BaseTask
|
|
worker SeqWorkFunc
|
|
}
|
|
|
|
func NewSeqTask(taskName string, ctx *jxcontext.Context, worker SeqWorkFunc, stepCount int, params ...interface{}) *SeqTask {
|
|
task := &SeqTask{
|
|
worker: worker,
|
|
}
|
|
task.Init(1, 1, false, params, taskName, ctx, stepCount, stepCount)
|
|
return task
|
|
}
|
|
|
|
func (task *SeqTask) Run() {
|
|
task.run(func() {
|
|
globals.SugarLogger.Debugf("SeqTask.Run %s", task.Name)
|
|
var taskErr error
|
|
var taskResult []interface{}
|
|
for i := 0; i < task.TotalItemCount; i++ {
|
|
select {
|
|
case <-task.quitChan:
|
|
goto EndFor
|
|
default:
|
|
}
|
|
result, err := task.callWorker(func() (retVal interface{}, err error) {
|
|
return task.worker(task, i, task.params...)
|
|
})
|
|
task.finishedOneJob(1, err)
|
|
if taskErr = err; taskErr != nil {
|
|
globals.SugarLogger.Infof("SeqTask.Run %s step:%d failed with error:%v", task.Name, i, err)
|
|
if !task.IsContinueWhenError {
|
|
break
|
|
}
|
|
task.locker.Lock()
|
|
task.detailErrMsgList = append(task.detailErrMsgList, err.Error())
|
|
task.locker.Unlock()
|
|
} else if result != nil {
|
|
taskResult = append(taskResult, utils.Interface2Slice(result)...)
|
|
}
|
|
}
|
|
EndFor:
|
|
task.locker.Lock()
|
|
if taskErr != nil { // 如果有错误，肯定就是失败了
|
|
task.Status = TaskStatusFailed
|
|
} else {
|
|
if task.FinishedJobCount < task.TotalJobCount {
|
|
taskErr = ErrTaskIsCanceled
|
|
task.Status = TaskStatusCanceled
|
|
} else {
|
|
task.Status = TaskStatusFinished
|
|
}
|
|
}
|
|
if taskErr != nil {
|
|
task.Err = NewTaskError(task.Name, taskErr)
|
|
} else {
|
|
task.Err = task.buildTaskErrFromDetail()
|
|
}
|
|
task.Result = taskResult
|
|
task.TerminatedAt = time.Now()
|
|
task.locker.Unlock()
|
|
globals.SugarLogger.Debugf("SeqTask.Run %s, result:%v, err:%v", task.Name, taskResult, task.Err)
|
|
})
|
|
}
|
|
|
|
func (t *SeqTask) AddChild(task ITask) ITask {
|
|
task.SetParent(t)
|
|
return t.BaseTask.AddChild(task)
|
|
}
|