liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- refactor jxcontent and tasksch (remove dependency from jxcontent to tasksch)

Browse Source 
- send dingding msg to user when async task finished
This commit is contained in:
gazebo
2019-03-22 15:04:28 +08:00
parent 0f5445020a
commit 25ac631c5c
31 changed files with 1527 additions and 1488 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
business/jxutils/jxcontext/jxcontext.go
View File

@@ -3,12 +3,10 @@ package jxcontext
import (
"errors"
"net/http"
"sync"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/auth2"
"git.rosy.net.cn/jx-callback/business/jxcallback/auth"
"git.rosy.net.cn/jx-callback/business/jxutils/tasksch"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego"
@@ -21,14 +19,12 @@ type IAuther interface {
}
type Context struct {
rootTask tasksch.ITask
token string
accessUUID string
userInfo IAuther //*auth.LoginInfo
w http.ResponseWriter
r *http.Request
mapData map[interface{}]interface{}
locker sync.RWMutex
}
const (
@@ -43,9 +39,8 @@ func init() {
AdminCtx = NewWithUserName(nil, model.AdminName, nil, nil)
}
func NewWithUserName(rootTask tasksch.ITask, userName string, w http.ResponseWriter, r *http.Request) (ctx *Context) {
func NewWithUserName(notUsed interface{}, userName string, w http.ResponseWriter, r *http.Request) (ctx *Context) {
ctx = &Context{
rootTask: rootTask,
token: userName,
w: w,
r: r,
@@ -55,9 +50,8 @@ func NewWithUserName(rootTask tasksch.ITask, userName string, w http.ResponseWri
return ctx
}
func New(rootTask tasksch.ITask, token string, w http.ResponseWriter, r *http.Request) (ctx *Context, errCode string, err error) {
func New(notUsed interface{}, token string, w http.ResponseWriter, r *http.Request) (ctx *Context, errCode string, err error) {
ctx = &Context{
rootTask: rootTask,
token: token,
w: w,
r: r,
@@ -141,34 +135,6 @@ func (ctx *Context) GetV2AuthInfo() (authInfo *auth2.AuthInfo, err error) {
return nil, auth2.ErrNeedV2Token
}
func (ctx *Context) GetRootTask() tasksch.ITask {
ctx.locker.RLock()
defer ctx.locker.RUnlock()
return ctx.rootTask
}
func (ctx *Context) SetTaskOrAddChild(task tasksch.ITask, parentTask tasksch.ITask) bool {
if parentTask != nil {
parentTask.AddChild(task)
} else {
parentTask = task
}
ctx.locker.RLock()
if ctx.rootTask == nil {
ctx.locker.RUnlock()
ctx.locker.Lock()
if ctx.rootTask == nil {
ctx.rootTask = parentTask
}
ctx.locker.Unlock()
return true
}
ctx.locker.RUnlock()
return false
}
func (ctx *Context) GetResponseWriter() http.ResponseWriter {
return ctx.w
}
@@ -176,17 +142,3 @@ func (ctx *Context) GetResponseWriter() http.ResponseWriter {
func (ctx *Context) GetRequest() *http.Request {
return ctx.r
}
func (ctx *Context) SetValue(key, value interface{}) {
ctx.locker.Lock()
defer ctx.locker.Unlock()
ctx.mapData[key] = value
}
func (ctx *Context) GetValue(key interface{}) interface{} {
ctx.locker.RLock()
defer ctx.locker.RUnlock()
return ctx.mapData[key]
}

24
business/jxutils/msg/msg.go Normal file
View File

@@ -0,0 +1,24 @@
package msg
import (
"git.rosy.net.cn/jx-callback/business/auth2"
"git.rosy.net.cn/jx-callback/business/auth2/authprovider/dingding"
"git.rosy.net.cn/jx-callback/globals"
"git.rosy.net.cn/jx-callback/globals/api"
)
func SendUserMessage(userID, title, content string) (err error) {
authList, err := auth2.GetUserBindAuthInfo(userID)
if err == nil {
for _, auth := range authList {
if auth.Type == dingding.AuthTypeStaff {
err = api.DingDingAPI.CorpAsyncSendSimple(auth.AuthID, content)
break
}
}
}
if err != nil {
globals.SugarLogger.Infof("SendUserMessage userID:%s, title:%s, content:%s failed with error:%v", userID, title, content, err)
}
return err
}

16
business/jxutils/tasksch/parallel_task.go
View File

@@ -6,6 +6,7 @@ import (
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
"git.rosy.net.cn/jx-callback/globals"
)
@@ -18,6 +19,8 @@ type WorkFunc func(task *ParallelTask, batchItemList []interface{}, params ...in
type ResultHandlerFunc func(taskName string, result []interface{}, err error)
type ParallelConfig struct {
// ParentTask ITask
// IsAsync bool
ParallelCount int
BatchSize int
IsContinueWhenError bool
@@ -41,9 +44,11 @@ var (
func NewParallelConfig() *ParallelConfig {
return &ParallelConfig{
// ParentTask: parentTask,
// IsAsync: false,
IsContinueWhenError: false,
ParallelCount: DefParallelCount,
BatchSize: 1,
IsContinueWhenError: false,
ResultHandler: nil,
}
}
@@ -63,12 +68,17 @@ func (c *ParallelConfig) SetIsContinueWhenError(isContinueWhenError bool) *Paral
return c
}
// func (c *ParallelConfig) SetIsAsync(isAsync bool) *ParallelConfig {
// c.IsAsync = isAsync
// return c
// }
func (c *ParallelConfig) SetResultHandler(resultHandler ResultHandlerFunc) *ParallelConfig {
c.ResultHandler = resultHandler
return c
}
func NewParallelTask(taskName string, config *ParallelConfig, userName string, worker WorkFunc, itemList interface{}, params ...interface{}) *ParallelTask {
func NewParallelTask(taskName string, config *ParallelConfig, ctx *jxcontext.Context, worker WorkFunc, itemList interface{}, params ...interface{}) *ParallelTask {
if config == nil {
config = NewParallelConfig()
}
@@ -91,7 +101,7 @@ func NewParallelTask(taskName string, config *ParallelConfig, userName string, w
worker: worker,
jobList: jobList,
}
task.Init(config.ParallelCount, config.BatchSize, config.IsContinueWhenError, params, taskName, userName, len(realItemList), jobListLen)
task.Init(config.ParallelCount, config.BatchSize, config.IsContinueWhenError, params, taskName, ctx, len(realItemList), jobListLen)
return task
}

70
business/jxutils/tasksch/parallel_task_test.go
View File

@@ -7,6 +7,7 @@ import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
)
func TestRunParallelTask(t *testing.T) {
@@ -14,16 +15,17 @@ func TestRunParallelTask(t *testing.T) {
for k := range itemList {
itemList[k] = k
}
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7), "autotest", func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
t.Logf("sleep %d seconds", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7), jxcontext.AdminCtx,
func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
t.Logf("sleep %d seconds", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task.Run()
result, err := task.GetResult(1 * time.Microsecond)
if err == nil || task.GetStatus() != TaskStatusWorking {
@@ -45,16 +47,17 @@ func TestCancelParallelTask(t *testing.T) {
for k := range itemList {
itemList[k] = k
}
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7), "autotest", func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
fmt.Printf("sleep %d seconds\n", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7), jxcontext.AdminCtx,
func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
fmt.Printf("sleep %d seconds\n", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task.Run()
// time.Sleep(time.Second * 6)
fmt.Printf("finishedItemCount:%d, finishedJobCount:%d\n", task.GetFinishedItemCount(), task.GetFinishedJobCount())
@@ -71,19 +74,20 @@ func TestRunParallelTaskPartialSuccess(t *testing.T) {
for k := range itemList {
itemList[k] = k
}
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7).SetIsContinueWhenError(true), "autotest", func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
t.Logf("sleep %d seconds", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
if rand.Intn(2) == 1 {
return nil, fmt.Errorf("test:%d", batchItemList[0].(int))
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task := NewParallelTask("test", NewParallelConfig().SetParallelCount(100).SetBatchSize(7).SetIsContinueWhenError(true), jxcontext.AdminCtx,
func(task *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
sleepSecond := rand.Intn(5)
t.Logf("sleep %d seconds", sleepSecond)
time.Sleep(time.Duration(sleepSecond) * time.Second)
retSlice := make([]string, len(batchItemList))
for k := range retSlice {
retSlice[k] = "hello:" + utils.Int2Str(batchItemList[k].(int)*2)
}
if rand.Intn(2) == 1 {
return nil, fmt.Errorf("test:%d", batchItemList[0].(int))
}
return retSlice, nil
}, itemList, "a", "b", 1, 2)
task.Run()
result, err := task.GetResult(1 * time.Microsecond)
if err == nil || task.GetStatus() != TaskStatusWorking {

5
business/jxutils/tasksch/sequence_task.go
View File

@@ -4,6 +4,7 @@ import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
"git.rosy.net.cn/jx-callback/globals"
)
@@ -19,11 +20,11 @@ type SeqTask struct {
worker SeqWorkFunc
}
func NewSeqTask(taskName string, userName string, worker SeqWorkFunc, stepCount int, params ...interface{}) *SeqTask {
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, userName, stepCount, stepCount)
task.Init(1, 1, false, params, taskName, ctx, stepCount, stepCount)
return task
}

59
business/jxutils/tasksch/sequence_task_test.go
View File

@@ -7,38 +7,41 @@ import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
)
func TestRunSeqTask(t *testing.T) {
var seqTask ITask
seqTask = NewSeqTask("TestSeqTask", "autotest", func(task *SeqTask, step int, params ...interface{}) (result interface{}, err error) {
switch step {
case 0:
fmt.Println("ONE")
task2 := NewParallelTask("hello", nil, "xjh", func(parallelTask *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
i := batchItemList[0].(int)
time.Sleep(2 * time.Second)
fmt.Println(i * 2)
return nil, nil
}, []int{1, 2, 3})
seqTask.AddChild(task2)
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
task2.Run()
case 1:
fmt.Println("TWO")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 2:
fmt.Println("THREE")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 3:
fmt.Println("FOUR")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 4:
fmt.Println("FIVE")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
}
return []string{"1"}, nil
}, 5)
seqTask = NewSeqTask("TestSeqTask", jxcontext.AdminCtx,
func(task *SeqTask, step int, params ...interface{}) (result interface{}, err error) {
switch step {
case 0:
fmt.Println("ONE")
task2 := NewParallelTask("hello", nil, jxcontext.AdminCtx,
func(parallelTask *ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
i := batchItemList[0].(int)
time.Sleep(2 * time.Second)
fmt.Println(i * 2)
return nil, nil
}, []int{1, 2, 3})
seqTask.AddChild(task2)
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
task2.Run()
case 1:
fmt.Println("TWO")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 2:
fmt.Println("THREE")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 3:
fmt.Println("FOUR")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
case 4:
fmt.Println("FIVE")
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
}
return []string{"1"}, nil
}, 5)
seqTask.Run()
time.Sleep(3 * time.Second)

39
business/jxutils/tasksch/task.go
View File

@@ -8,6 +8,8 @@ import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
"git.rosy.net.cn/jx-callback/business/jxutils/msg"
"git.rosy.net.cn/jx-callback/globals"
)
@@ -97,14 +99,17 @@ type BaseTask struct {
Children TaskList `json:"children"`
Err error `json:"err"`
detailErrMsgList []string `json:"-"`
finishChan chan int
C <-chan int `json:"-"`
detailErrMsgList []string
finishChan chan struct{}
C <-chan struct{} `json:"-"`
params []interface{}
quitChan chan int
locker sync.RWMutex
parent ITask
ctx *jxcontext.Context
isGetResultCalled bool
}
func (s TaskList) Len() int {
@@ -121,7 +126,7 @@ func (s TaskList) Swap(i, j int) {
s[j] = tmp
}
func (t *BaseTask) Init(parallelCount, batchSize int, isContinueWhenError bool, params []interface{}, name, userName string, totalItemCount, totalJobCount int) {
func (t *BaseTask) Init(parallelCount, batchSize int, isContinueWhenError bool, params []interface{}, name string, ctx *jxcontext.Context, totalItemCount, totalJobCount int) {
t.ID = utils.GetUUID()
t.ParallelCount = parallelCount
t.BatchSize = batchSize
@@ -129,13 +134,14 @@ func (t *BaseTask) Init(parallelCount, batchSize int, isContinueWhenError bool,
t.params = params
t.Name = name
t.CreatedAt = time.Now()
t.CreatedBy = userName
t.ctx = ctx
t.CreatedBy = ctx.GetUserName()
t.UpdatedAt = t.CreatedAt
t.TerminatedAt = utils.DefaultTimeValue
t.TotalItemCount = totalItemCount
t.TotalJobCount = totalJobCount
t.quitChan = make(chan int)
t.finishChan = make(chan int)
t.finishChan = make(chan struct{})
t.Status = TaskStatusWorking
t.C = t.finishChan
@@ -155,6 +161,7 @@ func (t *BaseTask) GetResult(duration time.Duration) (retVal []interface{}, err
timer := time.NewTimer(duration)
select {
case <-t.finishChan:
t.isGetResultCalled = true
timer.Stop()
return t.Result, t.Err
case <-timer.C:
@@ -249,6 +256,14 @@ func AddChild(parentTask ITask, task ITask) ITask {
return task
}
func HandleTask(task, parentTask ITask, isMangeIt bool) ITask {
AddChild(task, parentTask)
if parentTask == nil && isMangeIt {
ManageTask(task)
}
return task
}
/////////
func (t *BaseTask) MarshalJSON() ([]byte, error) {
@@ -280,6 +295,18 @@ func (t *BaseTask) run(taskHandler func()) {
}
close(t.finishChan)
time.Sleep(10 * time.Millisecond) // 等待GetResult中的isGetResultCalled赋值
if !t.isGetResultCalled && t.parent == nil && len(GetTasks(t.ID, TaskStatusBegin, TaskStatusEnd, 24, "")) == 0 {
if authInfo, err := t.ctx.GetV2AuthInfo(); err == nil { // 这里应该是不管登录类型，直接以可能的方式发消息
var content string
if t.Status == TaskStatusFinished {
content = fmt.Sprintf("你的异步任务[%s]执行成功完成", t.Name)
} else {
content = fmt.Sprintf("你的异步任务[%s]执行失败，%s", t.Name, t.Err.Error())
}
msg.SendUserMessage(authInfo.UserID, "异步任务完成", content)
}
}
})
}
}