You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
891 lines
29 KiB
891 lines
29 KiB
package service |
|
|
|
import ( |
|
"context" |
|
"fmt" |
|
"sort" |
|
"strconv" |
|
"strings" |
|
|
|
"github.com/jinzhu/gorm" |
|
"go-common/app/admin/main/aegis/model" |
|
"go-common/app/admin/main/aegis/model/net" |
|
"go-common/library/ecode" |
|
"go-common/library/log" |
|
"go-common/library/xstr" |
|
) |
|
|
|
//启动流程网之前的检查 |
|
func (s *Service) startNet(c context.Context, businessID int64, netID int64) (result *net.TriggerResult, err error) { |
|
var ( |
|
n *net.Net |
|
resultTokenPackage *net.TokenPackage |
|
) |
|
|
|
//网可用性 |
|
if n, err = s.netByID(c, netID); err != nil { |
|
log.Error("startNet s.netByID(%d) error(%v)", netID, err) |
|
return |
|
} |
|
if n == nil || !n.IsAvailable() || n.StartFlowID <= 0 { |
|
log.Error("startNet s.netByID(%d) not found/disabled/start_flow_id=0", netID) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
if n.BusinessID != businessID { |
|
log.Error("startNet s.netByID(%d) business(%d) != param business(%d)", netID, n.BusinessID, businessID) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
//初始节点的token集合 |
|
if resultTokenPackage, err = s.computeNewFlowPackage(c, n.StartFlowID, nil, nil, true); err != nil { |
|
log.Error("startNet s.computeNewFlowPackage(%d) error(%+v) netid(%d)", n.StartFlowID, err, netID) |
|
return |
|
} |
|
|
|
result = &net.TriggerResult{ |
|
NetID: netID, |
|
ResultToken: resultTokenPackage, |
|
NewFlowID: n.StartFlowID, |
|
From: model.LogFromStart, |
|
} |
|
return |
|
} |
|
|
|
/*FetchJumpFlowInfo .. |
|
* 跳流程详情页, 跳流程是单独db层面的更新 |
|
* 参数：flowid(可选) |
|
* 获取网下所有flow和变迁的所有操作项 |
|
*/ |
|
func (s *Service) FetchJumpFlowInfo(c context.Context, flowID int64) (res *net.JumpInfo, err error) { |
|
var ( |
|
flow *net.Flow |
|
runningNet int64 |
|
flowList []*net.Flow |
|
) |
|
|
|
res = &net.JumpInfo{ |
|
Flows: []*net.SimpleInfo{}, |
|
Operations: []*net.TranOperation{}, |
|
} |
|
if flow, err = s.flowByID(c, flowID); err != nil { |
|
log.Error("FetchJumpFlowInfo s.flowByID(%d) error(%v)", flowID, err) |
|
return |
|
} |
|
if flow == nil { |
|
log.Error("FetchJumpFlowInfo flow(%d) not found", flowID) |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
runningNet = flow.NetID |
|
if flowList, err = s.flowsByNet(c, runningNet); err != nil { |
|
log.Error("FetchJumpFlowInfo s.flowsByNet(%d) error(%v) flowid(%d)", runningNet, err, flowID) |
|
return |
|
} |
|
flowInfo := []*net.SimpleInfo{} |
|
for _, item := range flowList { |
|
flowInfo = append(flowInfo, &net.SimpleInfo{ |
|
ID: item.ID, |
|
ChName: item.ChName, |
|
}) |
|
} |
|
|
|
//网下的所有变迁可操作项 |
|
if res.Operations, err = s.tranOpersByNet(c, []int64{runningNet}, []int8{net.BindTypeTransition}); err != nil { |
|
log.Error("FetchJumpFlowInfo s.tranOpersByNet(%d) error(%v) flowid(%d)", runningNet, err, flowID) |
|
return |
|
} |
|
res.Flows = flowInfo |
|
return |
|
} |
|
|
|
/** |
|
* 跳流程提交：rid, oldflowid, newflowid, binds |
|
* 可以单独提供flowid,单独提供binds |
|
* 单独提供flowid时，不做流程流转 |
|
*/ |
|
func (s *Service) jumpFlow(c context.Context, tx *gorm.DB, rid int64, oldFlowID int64, newFlowID int64, binds []int64) (result *net.JumpFlowResult, err error) { |
|
var ( |
|
frs []*net.FlowResource |
|
oldFlow, newFlow *net.Flow |
|
triggerResult *net.TriggerResult |
|
bindList []*net.TokenBind |
|
submitPackage *net.TokenPackage |
|
) |
|
|
|
log.Info("jumpFlow before rid(%d) oldflowid(%d) newflowid(%d) binds(%v)", rid, oldFlowID, newFlowID, binds) |
|
if rid <= 0 || oldFlowID <= 0 || (len(binds) == 0 && newFlowID <= 0) { |
|
err = ecode.RequestErr |
|
return |
|
} |
|
//检查资源是否运行在该节点 |
|
if frs, err = s.gorm.FRByUniques(c, []int64{rid}, []int64{oldFlowID}, true); err != nil { |
|
log.Error("jumpFlow s.gorm.FRByUniques(%d,%d) error(%v)", rid, oldFlowID, err) |
|
return |
|
} |
|
if len(frs) == 0 { |
|
log.Error("jumpFlow rid(%d) not running at oldflowid(%d)", rid, oldFlowID) |
|
err = ecode.AegisNotRunInFlow |
|
return |
|
} |
|
if oldFlow, err = s.flowByID(c, oldFlowID); err != nil { |
|
log.Error("jumpFlow s.flowByID(%d) error(%v) rid(%d)", oldFlowID, err, rid) |
|
return |
|
} |
|
if oldFlow == nil { |
|
log.Error("jumpFlow oldflowid(%d) not found, rid(%d)", oldFlowID, rid) |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
|
|
/** |
|
* jumpFlow 指定一批资源，更新为相同现状 |
|
* 不需要前端指定现状flowid |
|
* 资源现状从单线跳到多线程的某个节点时----由运营决定是到sync-split节点还是并发分支上的某个节点------fr数目从1变成1/还是1变成n |
|
* 资源现状从多线程跳到某个节点时----节点在并发分支上，如何确定现状flow和新节点都在并发分支上----fr数目从1变成1 |
|
* ----节点在单线上，结束所有运行中都并发现状，加一个单线分支---fr数目从n变成1 |
|
*/ |
|
if len(binds) > 0 { |
|
if bindList, err = s.tokenBinds(c, binds, true); err != nil { |
|
log.Error("jumpFlow s.tokenBinds(%v) error(%v) rid(%d)", binds, err, rid) |
|
return |
|
} |
|
if len(bindList) == 0 { |
|
log.Error("jumpFlow binds(%v) not found, rid(%d)", binds, rid) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
if submitPackage, err = s.newTokenPackage(c, bindList, true); err != nil { |
|
log.Error("jumpFlow s.newTokenPackage(%v) error(%v) rid(%d)", binds, err, rid) |
|
return |
|
} |
|
} |
|
|
|
if newFlowID > 0 && newFlowID != oldFlowID { |
|
if newFlow, err = s.flowByID(c, newFlowID); err != nil { |
|
log.Error("jumpFlow newflow(%d) error(%v) rid(%d)", newFlowID, err, rid) |
|
return |
|
} |
|
if newFlow == nil { |
|
log.Error("jumpFlow newflow(%d) not found, rid(%d)", newFlowID, rid) |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
if oldFlow.NetID != newFlow.NetID { |
|
log.Error("jumpFlow rid(%d) run at oldflowid(%d)/net(%d), can't jump to newflowid(%d)/net(%d)", rid, oldFlowID, oldFlow.NetID, newFlowID, newFlow.NetID) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
triggerResult = &net.TriggerResult{ |
|
RID: rid, |
|
NetID: oldFlow.NetID, |
|
NewFlowID: newFlowID, |
|
OldFlowID: strconv.FormatInt(oldFlowID, 10), |
|
SubmitToken: submitPackage, |
|
ResultToken: submitPackage, |
|
} |
|
|
|
if err = s.reachNewFlowDB(c, tx, triggerResult); err != nil { |
|
log.Error("jumpFlow s.reachNewFlowDB error(%v) triggerresult(%+v)", err, triggerResult) |
|
return |
|
} |
|
} else { |
|
newFlowID = oldFlowID |
|
} |
|
|
|
result = &net.JumpFlowResult{ |
|
RID: rid, |
|
NetID: oldFlow.NetID, |
|
SubmitToken: submitPackage, |
|
ResultToken: submitPackage, |
|
NewFlowID: newFlowID, |
|
OldFlowID: strconv.FormatInt(oldFlowID, 10), |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) afterJumpFlow(c context.Context, res *net.JumpFlowResult, bizid int64) (err error) { |
|
trigger := &net.TriggerResult{ |
|
RID: res.RID, |
|
NetID: res.NetID, |
|
SubmitToken: res.SubmitToken, |
|
ResultToken: res.ResultToken, |
|
NewFlowID: res.NewFlowID, |
|
OldFlowID: res.OldFlowID, |
|
From: model.LogFromJump, |
|
} |
|
|
|
s.afterReachNewFlow(c, trigger, bizid) |
|
return |
|
} |
|
|
|
/** |
|
* 批量审核提交--单个遍历处理 |
|
* 参数：business, rid, binds |
|
* flowid(由rid反查现状) |
|
* transition_id（由flow_id查到的下游transition & token_bind集合 与 token_id的交集决定），非叶子节点若找不到则报错 |
|
* 找到了,则计算新flow & result_token |
|
* 计算result_token规则： |
|
* 变迁trigger=人工，必须提供binds; |
|
* 变迁trigger=其他，binds可以为空 & （新flow要么有绑定token，要么有向线上output不为空） |
|
*/ |
|
func (s *Service) computeBatchTriggerResult(c context.Context, businessID int64, rid int64, bindID []int64) (result *net.TriggerResult, err error) { |
|
var ( |
|
oldFlow *net.Flow |
|
) |
|
|
|
log.Info("computeBatchTriggerResult start business(%d) rid(%d) bindID(%+v)", businessID, rid, bindID) |
|
|
|
//反查指定资源所运行的节点 |
|
if oldFlow, _, err = s.flowsByRunning(c, rid, businessID, 0, true); err != nil { |
|
return |
|
} |
|
|
|
if result, err = s.computeResult(c, rid, oldFlow, bindID, true); err != nil { |
|
log.Error("computeBatchTriggerResult s.computeResult error(%v) rid(%d) businessid(%d) bindID(%v) oldflow(%+v)", err, rid, businessID, bindID, oldFlow) |
|
return |
|
} |
|
result.From = model.LogFromBatch |
|
log.Info("computeBatchTriggerResult end business(%d) rid(%d) bindID(%+v) result(%+v)", businessID, rid, bindID, result) |
|
return |
|
} |
|
|
|
/** |
|
* 详情页审核提交 |
|
* 参数：rid, flowid, binds |
|
* 当flowid=最后一个，不流转--binds可能为多个; 其余情况，均正常流转到下一个flow--binds只有一个 |
|
* 计算result_token规则： |
|
* 变迁trigger=人工，必须提供binds; |
|
* 变迁trigger=其他，binds可以为空 & （新flow要么有绑定token，要么有向线上output不为空） |
|
*/ |
|
func (s *Service) computeTriggerResult(c context.Context, rid int64, oldFlowID int64, binds []int64) (result *net.TriggerResult, err error) { |
|
var ( |
|
oldFlow *net.Flow |
|
frs []*net.FlowResource |
|
) |
|
|
|
log.Info("computeTriggerResult start rid(%d) oldflow(%d) binds(%+v)", rid, oldFlowID, binds) |
|
if rid <= 0 || oldFlowID <= 0 || len(binds) == 0 { |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if frs, err = s.gorm.FRByUniques(c, []int64{rid}, []int64{oldFlowID}, true); err != nil { |
|
log.Error("computeTriggerResult s.gorm.FRByUniques(%d) error(%v) rid(%d) binds(%v)", oldFlowID, err, rid, binds) |
|
return |
|
} |
|
if len(frs) == 0 { |
|
log.Error("computeTriggerResult rid(%d) not running at oldflow(%d) binds(%v)", rid, oldFlowID, binds) |
|
err = ecode.AegisNotRunInFlow |
|
return |
|
} |
|
if oldFlow, err = s.flowByID(c, oldFlowID); err != nil { |
|
log.Error("computeTriggerResult s.flowByID(%d) error(%v) rid(%d) binds(%v)", oldFlowID, err, rid, binds) |
|
return |
|
} |
|
if oldFlow == nil { |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
|
|
if result, err = s.computeResult(c, rid, oldFlow, binds, false); err != nil { |
|
log.Error("computeTriggerResult s.computeResult error(%v) binds(%v) flow(%+v) rid(%d)", err, binds, oldFlow, rid) |
|
return |
|
} |
|
result.From = model.LogFromSingle |
|
log.Info("computeTriggerResult end rid(%d) oldflow(%d) binds(%+v) result(%+v)", rid, oldFlowID, binds, result) |
|
return |
|
} |
|
|
|
func (s *Service) computeResult(c context.Context, rid int64, flow *net.Flow, bindID []int64, fromBatch bool) (result *net.TriggerResult, err error) { |
|
var ( |
|
hitAudit, hitHelper bool //审核流转和非流转操作是互斥的，不能同时提交 |
|
dirs []*net.Direction |
|
isLast bool |
|
triggerBind = []*net.TokenBind{} |
|
binds []*net.TokenBind |
|
trans []*net.Transition |
|
triggerTranID, newFlowID int64 |
|
submitTokenPackage, resultTokenPackage *net.TokenPackage |
|
resultDir *net.Direction |
|
) |
|
|
|
//非叶子节点只能有一个bind |
|
if rid == 0 || flow == nil { |
|
log.Error("computeResult rid(%d)=0/flow(%+v)=ni", rid, flow) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if binds, err = s.tokenBinds(c, bindID, true); err != nil { |
|
log.Error("computeBatchTriggerResult s.tokenBinds(%+v) error(%v)", bindID, err) |
|
return |
|
} |
|
tranID := []int64{} |
|
for _, item := range binds { |
|
if item.IsBatch() != fromBatch { |
|
continue |
|
} |
|
|
|
tranID = append(tranID, item.ElementID) |
|
if (fromBatch && item.Type == net.BindTypeTranBatch) || (!fromBatch && item.Type == net.BindTypeTransition) { |
|
hitAudit = true |
|
} |
|
if (fromBatch && item.Type == net.BindTypeTranHelpBatch) || (!fromBatch && item.Type == net.BindTypeTranHelp) { |
|
hitHelper = true |
|
} |
|
} |
|
if len(binds) == 0 || (hitAudit && hitHelper) { |
|
log.Error("computeBatchTriggerResult binds(%v) not found/both hit audit && helper are not allowed", binds) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if hitAudit { |
|
//同网下、同flow下游的变迁过滤, 非叶子节点必须命中一个变迁 |
|
if dirs, err = s.dirByFlow(c, []int64{flow.ID}, net.DirInput); err != nil { |
|
log.Error("computeResult s.dirByFlow(%d) error(%v)", flow.ID, err) |
|
return |
|
} |
|
isLast = len(dirs) == 0 //是否为叶子节点 |
|
if !isLast { |
|
tranID = []int64{} |
|
for _, item := range dirs { |
|
tranID = append(tranID, item.TransitionID) |
|
} |
|
} |
|
} |
|
|
|
if trans, err = s.transitions(c, tranID, true); err != nil { |
|
log.Error("computeResult s.transitions(%v) error(%v) rid(%d) flow(%v)", tranID, err, rid, flow) |
|
return |
|
} |
|
tranMap := map[int64]*net.Transition{} |
|
for _, item := range trans { |
|
tranMap[item.ID] = item |
|
} |
|
triggerBindID := []int64{} |
|
triggerTrans := []int64{} |
|
for _, item := range binds { |
|
if tranMap[item.ElementID] == nil || tranMap[item.ElementID].Trigger != net.TriggerManual { |
|
continue |
|
} |
|
|
|
//叶子节点=同网过滤|非叶子节点=同flow下游变迁过滤 |
|
if (isLast && tranMap[item.ElementID].NetID == flow.NetID) || !isLast { |
|
triggerBind = append(triggerBind, item) |
|
triggerTrans = append(triggerTrans, item.ElementID) |
|
triggerBindID = append(triggerBindID, item.ID) |
|
} |
|
} |
|
if len(triggerBind) == 0 || (!isLast && len(triggerBind) > 1) { |
|
log.Error("computeResult no triggered(%+v)/non-leaf-flow triggered >2, rid(%d) flow(%+v)", triggerBindID, rid, flow) |
|
err = ecode.AegisNotTriggerFlow |
|
return |
|
} |
|
|
|
//计算提交的变迁 & 提交令牌集合 |
|
if submitTokenPackage, err = s.newTokenPackage(c, triggerBind, true); err != nil { |
|
log.Error("computeResult s.newTokenPackage error(%v)", err) |
|
return |
|
} |
|
|
|
if hitHelper || isLast { |
|
//不流转的情况 |
|
newFlowID = flow.ID |
|
resultTokenPackage = submitTokenPackage |
|
} else { |
|
//中间节点找到下游新节点 |
|
triggerTranID = triggerTrans[0] |
|
if resultDir, err = s.fetchTranNextEnableDirs(c, triggerTranID); err != nil { |
|
log.Error("computeResult s.fetchTranNextEnableDirs(%d) error(%v)", triggerTranID, err) |
|
return |
|
} |
|
|
|
newFlowID = resultDir.FlowID |
|
if resultTokenPackage, err = s.computeNewFlowPackage(c, newFlowID, resultDir, submitTokenPackage, false); err != nil { |
|
log.Error("computeResult s.computeNewFlowPackage error(%v)", err) |
|
return |
|
} |
|
} |
|
|
|
log.Info("submitpk(%+v) result(%+v)", submitTokenPackage, resultTokenPackage) |
|
result = &net.TriggerResult{ |
|
RID: rid, |
|
NetID: flow.NetID, |
|
SubmitToken: submitTokenPackage, |
|
ResultToken: resultTokenPackage, |
|
NewFlowID: newFlowID, |
|
OldFlowID: strconv.FormatInt(flow.ID, 10), |
|
TransitionID: triggerTrans, |
|
} |
|
log.Info("computeResult end islast(%v) frombatch(%v) result(%+v)", isLast, fromBatch, result) |
|
return |
|
} |
|
|
|
/** |
|
* 资源审核详情页，获取运行资源所在网的节点下的所有可允许下游可操作项，去掉批量可操作项: |
|
* 若为网中的最后一个flow, 获取网中所有变迁的可操作项(bind级别去重)--提交时只改token，不改flow |
|
* 若为中间节点：只有一个（非并发），直接获取变迁的可操作项; 并发，需指定哪个变迁(先抛错) |
|
* 前提：同一rid只在同一业务下的一个net下运行 |
|
* 任务列表进入详情页参数：rid,businessid,netid(可选) |
|
*/ |
|
func (s *Service) fetchResourceTranInfo(c context.Context, rid int64, businessID int64, netID int64) (result *net.TransitionInfo, err error) { |
|
var ( |
|
runningFlow *net.Flow |
|
runningNet int64 |
|
) |
|
if rid <= 0 || businessID <= 0 { |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if runningFlow, runningNet, err = s.flowsByRunning(c, rid, businessID, netID, false); err != nil { |
|
return |
|
} |
|
flowID := runningFlow.ID |
|
|
|
if result, err = s.fetchTaskTranInfo(c, rid, flowID, runningNet); err != nil { |
|
log.Error("fetchResourceTranInfo s.fetchTaskTranInfo error(%v) rid(%d) businessid(%d) netid(%d) runningnet(%d)", err, rid, businessID, netID, runningNet) |
|
} |
|
return |
|
} |
|
|
|
/** |
|
* 任务详情页，获取flowid下的所有可允许变迁的可操作项，去掉批量可操作项: |
|
* 若为网中的最后一个flow, 获取网中所有变迁的可操作项(bind级别去重)--提交时只改token，不改flow |
|
* 若为中间节点：只有一个（非并发），直接获取变迁的可操作项;并发，需指定哪个变迁(先抛错) |
|
* 前提：同一rid只在同一业务下的一个net下运行 |
|
* 资源列表进入详情页参数：rid, flowid, netid(可选) |
|
*/ |
|
func (s *Service) fetchTaskTranInfo(c context.Context, rid int64, flowID int64, netID int64) (result *net.TransitionInfo, err error) { |
|
var ( |
|
transitionID []int64 |
|
flow *net.Flow |
|
enableDir []*net.Direction |
|
trans []*net.Transition |
|
) |
|
|
|
if rid <= 0 || flowID <= 0 { |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if enableDir, err = s.fetchFlowNextEnableDirs(c, flowID); err != nil { |
|
log.Error("fetchTaskTranInfo s.fetchFlowNextEnableDirs(%d) error(%v)", flowID, err) |
|
return |
|
} |
|
//作为分支的叶子节点,获取整个网的所有可操作项 |
|
if len(enableDir) == 0 { |
|
if netID <= 0 { |
|
if flow, err = s.flowByID(c, flowID); err != nil { |
|
log.Error("fetchTaskTranInfo s.flowByID error(%v) rid(%d) flowid(%d)", err, rid, flowID) |
|
return |
|
} |
|
if flow == nil { |
|
log.Error("fetchTaskTranInfo flow(%d) not found rid(%d) ", flowID, rid) |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
netID = flow.NetID |
|
} |
|
|
|
if transitionID, err = s.tranIDByNet(c, []int64{netID}, true, true); err != nil { |
|
log.Error("fetchTaskTranInfo s.tranIDByNet(%d) error(%v) rid(%d) flowid(%d)", netID, err, rid, flowID) |
|
return |
|
} |
|
} else { |
|
transitionID = []int64{} |
|
for _, item := range enableDir { |
|
transitionID = append(transitionID, item.TransitionID) |
|
} |
|
if trans, err = s.transitions(c, transitionID, false); err != nil { |
|
log.Error("fetchTaskTranInfo s.transitions(%v) error(%v) rid(%d) flowid(%d)", transitionID, err, rid, flowID) |
|
return |
|
} |
|
transitionID = []int64{} |
|
for _, item := range trans { |
|
if item.Trigger == net.TriggerManual { |
|
transitionID = append(transitionID, item.ID) |
|
} |
|
} |
|
} |
|
|
|
result = &net.TransitionInfo{ |
|
RID: rid, |
|
FlowID: flowID, |
|
} |
|
if len(transitionID) == 0 { |
|
return |
|
} |
|
|
|
if result.Operations, err = s.tranOpers(c, transitionID, []int8{net.BindTypeTransition, net.BindTypeTranHelp}); err != nil { |
|
log.Error("fetchTaskTranInfo s.tranOpers(%v) error(%v) rid(%d) flowid(%d)", transitionID, err, rid, flowID) |
|
return |
|
} |
|
return |
|
} |
|
|
|
/** |
|
* 全部资源列表页，获取所有变迁的所有批量操作项 |
|
* 参数：businessid,netid(选填) |
|
*/ |
|
func (s *Service) fetchBatchOperations(c context.Context, businessID int64, netID int64) (operations []*net.TranOperation, err error) { |
|
var ( |
|
netIDList []int64 |
|
) |
|
|
|
operations = []*net.TranOperation{} |
|
if netID > 0 { |
|
netIDList = []int64{netID} |
|
} else { |
|
//查询business_id下的所有net |
|
if netIDList, err = s.netIDByBusiness(c, businessID); err != nil { |
|
log.Error("fetchBatchOperations s.netIDByBusiness error(%v) businessid(%d) netid(%d)", err, businessID, netID) |
|
return |
|
} |
|
if len(netIDList) == 0 { |
|
log.Error("fetchBatchOperations business(%d) no net", businessID) |
|
return |
|
} |
|
} |
|
|
|
//查询所有net下的所有变迁可操作项 |
|
if operations, err = s.tranOpersByNet(c, netIDList, []int8{net.BindTypeTranBatch, net.BindTypeTranHelpBatch}); err != nil { |
|
log.Error("fetchBatchOperations s.tranOpersByNet(%v) error(%v) businessid(%d) netid(%d)", netIDList, err, businessID, netID) |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) tranOpersByNet(c context.Context, netIDs []int64, bindTp []int8) (opers []*net.TranOperation, err error) { |
|
var ( |
|
transitionID = []int64{} |
|
) |
|
|
|
opers = []*net.TranOperation{} |
|
if len(netIDs) == 0 { |
|
return |
|
} |
|
if transitionID, err = s.tranIDByNet(c, netIDs, true, true); err != nil { |
|
log.Error("tranOpersByNet s.tranIDByNet(%v) error(%v) isBatch(%v)", netIDs, err, bindTp) |
|
return |
|
} |
|
if len(transitionID) == 0 { |
|
return |
|
} |
|
if opers, err = s.tranOpers(c, transitionID, bindTp); err != nil { |
|
log.Error("tranOpersByNet s.tranOpers(%v) error(%v) netid(%v) bindtp(%v)", transitionID, err, netIDs, bindTp) |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) tranOpers(c context.Context, tranID []int64, bindTp []int8) (opers []*net.TranOperation, err error) { |
|
var ( |
|
binds = []*net.TokenBind{} |
|
) |
|
|
|
opers = []*net.TranOperation{} |
|
if binds, err = s.tokenBindByElement(c, tranID, bindTp); err != nil { |
|
log.Error("tranOpers s.tokenBindByElement error(%v) tranid(%v) bindtp(%v)", err, tranID, bindTp) |
|
return |
|
} |
|
//bind.token_id维度去重 + bind.id维度排序 |
|
tokenIDBind := map[string][]int64{} |
|
unique := map[string]*net.TranOperation{} |
|
for _, item := range binds { |
|
if _, exist := tokenIDBind[item.TokenID]; !exist { |
|
tokenIDBind[item.TokenID] = []int64{item.ID} |
|
unique[item.TokenID] = &net.TranOperation{ |
|
ChName: item.ChName, |
|
} |
|
continue |
|
} |
|
tokenIDBind[item.TokenID] = append(tokenIDBind[item.TokenID], item.ID) |
|
} |
|
|
|
for uniqueTokenID, item := range unique { |
|
item.BindIDList = xstr.JoinInts(tokenIDBind[uniqueTokenID]) |
|
opers = append(opers, item) |
|
} |
|
sort.Sort(net.TranOperationArr(opers)) |
|
return |
|
} |
|
|
|
/** |
|
* 计算新节点的token集合 |
|
* 启动流程节点，只提供newflowID, 找绑定的token |
|
* 流转过程中，根据bindIDs + 触发变迁的可允许有向线(变迁id + newflowid)计算新flow的结果： |
|
* 1. 若有静态绑定token,返回绑定的 |
|
* 2. 若没静态绑定，且output="",返回bindID |
|
* 3. 若没静态绑定，且output!="",解析output计算---计算可能涉及到触发变迁/上一个flowid |
|
* 4. 其他情况，为配置错误，应该在配置时避免---todo |
|
* 5. start临时支持无令牌绑定情况 |
|
*/ |
|
func (s *Service) computeNewFlowPackage(c context.Context, newFlowID int64, fromDir *net.Direction, submitPackage *net.TokenPackage, start bool) (resultTokens *net.TokenPackage, err error) { |
|
var ( |
|
binds []*net.TokenBind |
|
) |
|
|
|
//flow绑定了tokens,直接返回 |
|
if binds, err = s.tokenBindByElement(c, []int64{newFlowID}, []int8{net.BindTypeFlow}); err != nil { |
|
log.Error("computeNewFlowPackage s.tokenBindByElement error(%v) newflowid(%d) fromdir(%v) submit(%+v)", err, newFlowID, fromDir, submitPackage) |
|
return |
|
} |
|
if len(binds) > 0 { |
|
if resultTokens, err = s.newTokenPackage(c, binds, false); err != nil { |
|
log.Error("computeNewFlowPackage s.newTokenPackage error(%v) newflowid(%d) fromdir(%v) submit(%+v)", err, newFlowID, fromDir, submitPackage) |
|
} |
|
return |
|
} |
|
if start { |
|
return |
|
} |
|
|
|
//没绑定的，通过有向线的output计算 |
|
if fromDir == nil { |
|
err = ecode.AegisFlowNoFromDir |
|
log.Error("computeNewFlowPackage newflowid(%d) has no tokens & fromdir, submit(%+v)", newFlowID, submitPackage) |
|
return |
|
} |
|
//output为空，根据提交内容 |
|
if fromDir.Output == "" { |
|
resultTokens = submitPackage |
|
return |
|
} |
|
|
|
//todo--compute output，解析与prevFlowID + rid + submitTokens相关,return {{"rids":[],"tokens":[]}} --version2 |
|
|
|
return |
|
} |
|
|
|
/** |
|
* 计算指定令牌关联的打包形式，包括：各类型的值+关联中文名+关联对应的令牌 |
|
* sametokenid=true，需检查binds对应的tokenid是否一致 |
|
*/ |
|
func (s *Service) newTokenPackage(c context.Context, binds []*net.TokenBind, sameTokenID bool) (res *net.TokenPackage, err error) { |
|
var ( |
|
tokenIDList []int64 |
|
tokens []*net.Token |
|
value interface{} |
|
sameTokenName string |
|
hitAudit bool |
|
) |
|
|
|
tokenIDStr := "" |
|
tokenID := "" |
|
if sameTokenID { |
|
tokenID = binds[0].TokenID |
|
sameTokenName = binds[0].ChName |
|
} |
|
for _, item := range binds { |
|
if sameTokenID && tokenID != item.TokenID { |
|
err = ecode.RequestErr |
|
log.Error("newTokenPackage binds diff token(%s!=%s) sameTokenID(%v)", tokenID, item.TokenID, sameTokenID) |
|
return |
|
} |
|
|
|
if item.Type == net.BindTypeTransition || item.Type == net.BindTypeTranBatch { |
|
hitAudit = true |
|
} |
|
tokenIDStr = tokenIDStr + "," + item.TokenID |
|
} |
|
tokenIDStr = strings.TrimLeft(tokenIDStr, ",") |
|
|
|
if tokenIDList, err = xstr.SplitInts(tokenIDStr); err != nil { |
|
log.Error("newTokenPackage xstr.SplitInts(%s) error(%v) sameTokenID(%v)", tokenIDStr, err, sameTokenID) |
|
return |
|
} |
|
|
|
if tokens, err = s.tokens(c, tokenIDList); err != nil { |
|
log.Error("newTokenPackage s.tokens(%v) error(%v) sameTokenID(%v)", tokenIDStr, err, sameTokenID) |
|
return |
|
} |
|
if len(tokens) == 0 { |
|
log.Error("newTokenPackage tokens(%s) not found sameTokenID(%v)", tokenIDStr, sameTokenID) |
|
err = ecode.AegisTokenNotFound |
|
return |
|
} |
|
|
|
values := map[string]interface{}{} |
|
chName := sameTokenName |
|
for _, item := range tokens { |
|
if value, err = item.FormatValue(); err != nil { |
|
log.Error("NewTokenPackage item.FormatValue(%+v) error(%v) sameTokenID(%v)", item, err, sameTokenID) |
|
return |
|
} |
|
values[item.Name] = value |
|
if sameTokenName == "" { |
|
chName = chName + item.ChName |
|
} |
|
} |
|
|
|
res = &net.TokenPackage{ |
|
Values: values, |
|
TokenIDList: tokenIDList, |
|
ChName: chName, |
|
HitAudit: hitAudit, |
|
} |
|
return |
|
} |
|
|
|
/** |
|
* flowsByRunning 指定业务或netid，获取资源的运行节点, 只在一个节点上运行 |
|
* rid反查现状得到flows,netid或businessid做过滤 |
|
* businessID可选，netID可选，2者必须提供一个 |
|
* onlyRunning 只过滤正常运行的资源现状 |
|
*/ |
|
func (s *Service) flowsByRunning(c context.Context, rid int64, businessID int64, netID int64, onlyRunning bool) (runningFlow *net.Flow, runningNetID int64, err error) { |
|
var ( |
|
n *net.Net |
|
nets []int64 |
|
frs []*net.FlowResource |
|
) |
|
if rid <= 0 || (businessID <= 0 && netID == 0) { |
|
err = ecode.RequestErr |
|
log.Error("flowsByRunning rid(%d)/businessid(%d)+netid(%d) are empty, onlyrunning(%v)", rid, businessID, netID, onlyRunning) |
|
return |
|
} |
|
|
|
if netID > 0 { |
|
if n, err = s.netByID(c, netID); err != nil { |
|
log.Error("flowsByRunning s.netByID(%d) error(%v) rid(%d) businessid(%d) onlyrunning(%v)", netID, err, rid, businessID, onlyRunning) |
|
return |
|
} |
|
if n == nil || n.BusinessID != businessID { |
|
log.Error("flowsByRunning net(%d) in business(%d) not found, rid(%d) onlyrunning(%v)", netID, businessID, rid, onlyRunning) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
nets = []int64{netID} |
|
} else { |
|
if nets, err = s.netIDByBusiness(c, businessID); err != nil { |
|
log.Error("flowsByRunning s.netIDByBusiness(%d) error(%v) rid(%d) onlyrunning(%v)", businessID, err, rid, onlyRunning) |
|
return |
|
} |
|
} |
|
if frs, err = s.gorm.FRByNetRID(c, nets, []int64{rid}, onlyRunning); err != nil { |
|
log.Error("flowsByRunning s.gorm.FRByNetRID(%d) error(%v) businessid(%d) netid(%d)", rid, err, businessID, netID) |
|
return |
|
} |
|
if len(frs) == 0 { |
|
log.Error("flowsByRunning rid(%d) not running in business(%d)/netid(%d) onlyrunning(%v)", rid, businessID, netID, onlyRunning) |
|
err = ecode.AegisNotRunInRange |
|
return |
|
} |
|
|
|
flowID := []int64{} |
|
for _, item := range frs { |
|
if runningNetID == 0 { |
|
runningNetID = item.NetID |
|
} else if item.NetID != runningNetID { |
|
log.Error("flowsByRunning rid(%d) running is both net(%d) & net(%d), business(%d), onlyrunning(%v)", rid, runningNetID, item.NetID, businessID, onlyRunning) |
|
err = ecode.AegisRunInDiffNet |
|
return |
|
} |
|
|
|
flowID = append(flowID, item.FlowID) |
|
} |
|
if len(flowID) > 1 { |
|
log.Error("flowsByRunning rid(%d) running in flows(%+v)>=2, businessid(%d)/netid(%d) runningnet(%d), onlyrunning(%v)", rid, flowID, businessID, netID, runningNetID, onlyRunning) |
|
err = ecode.AegisNotRunInFlow |
|
return |
|
} |
|
|
|
if runningFlow, err = s.flowByID(c, flowID[0]); err != nil { |
|
log.Error("flowsByRunning s.flowByID(%d) error(%v), rid(%d) businessid(%d) netid(%d), onlyrunning(%v)", flowID[0], err, rid, businessID, netID, onlyRunning) |
|
return |
|
} |
|
if runningFlow == nil { |
|
log.Error("flowsByRunning rid(%d) running in flows(%d) not found, businessid(%d) netid(%d) onlyrunnning(%v)", rid, flowID[0], businessID, netID, onlyRunning) |
|
err = ecode.AegisFlowNotFound |
|
return |
|
} |
|
return |
|
} |
|
|
|
/** |
|
* 流转到新节点且已更新现状表后，所需做的处理 |
|
* 1. 现状流转日志 |
|
* 2. 下游变迁处理，比如：人工变迁是否需要分发 |
|
*/ |
|
func (s *Service) afterReachNewFlow(c context.Context, pm *net.TriggerResult, bizid int64) (err error) { |
|
s.sendNetTriggerLog(c, pm) |
|
err = s.prepareBeforeTrigger(c, []int64{pm.RID}, pm.NewFlowID, bizid) |
|
return |
|
} |
|
|
|
/** |
|
* 流转到达新节点，更新db操作，需在新节点计算token之后执行 |
|
* 1. 现状表更新 |
|
*/ |
|
func (s *Service) reachNewFlowDB(c context.Context, tx *gorm.DB, pm *net.TriggerResult) (err error) { |
|
var ( |
|
flow *net.Flow |
|
) |
|
if pm.NetID <= 0 || pm.RID <= 0 || pm.NewFlowID <= 0 { |
|
log.Error("reachNewFlowDB params error(%+v)", pm) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
//新节点与netid的同网检查 |
|
if flow, err = s.flowByID(c, pm.NewFlowID); err != nil { |
|
log.Error("reachNewFlowDB s.flowByID error(%v) params(%+v)", err, pm) |
|
return |
|
} |
|
if flow == nil || flow.NetID != pm.NetID { |
|
log.Error("reachNewFlowDB s.flowByID not found/not same net, params(%+v), flow(%+v)", pm, flow) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
if err = s.updateFlowResources(c, tx, pm.NetID, pm.RID, pm.NewFlowID); err != nil { |
|
log.Error("reachNewFlowDB s.changeFlowResource error(%v) params(%+v)", err, pm) |
|
return |
|
} |
|
//todo--更新现状token情况 |
|
|
|
return |
|
} |
|
|
|
//取消指定资源的所有运行中流程 |
|
func (s *Service) cancelNet(c context.Context, tx *gorm.DB, rids []int64) (res map[int64]string, err error) { |
|
var ( |
|
frs []*net.FlowResource |
|
) |
|
|
|
if len(rids) == 0 { |
|
return |
|
} |
|
res = map[int64]string{} |
|
if frs, err = s.gorm.FRByUniques(c, rids, nil, true); err != nil { |
|
log.Error("cancelNet s.gorm.FRByUniques error(%v) rids(%+v)", err, rids) |
|
return |
|
} |
|
if len(frs) == 0 { |
|
return |
|
} |
|
if err = s.gorm.CancelFlowResource(c, tx, rids); err != nil { |
|
log.Error("cancelNet s.gorm.CancelFlowResource error(%+v) rids(%+v)", err, rids) |
|
return |
|
} |
|
|
|
//添加日志 |
|
trigger := &net.TriggerResult{ |
|
From: model.LogFromCancle, |
|
} |
|
for _, item := range frs { |
|
pref := res[item.RID] |
|
if pref != "" { |
|
pref = "," + pref |
|
} |
|
res[item.RID] = fmt.Sprintf("%s%d", pref, item.FlowID) |
|
trigger.RID = item.RID |
|
trigger.NetID = item.NetID |
|
trigger.OldFlowID = strconv.FormatInt(item.FlowID, 10) |
|
trigger.NewFlowID = item.FlowID |
|
s.sendNetTriggerLog(c, trigger) |
|
} |
|
return |
|
}
|
|
|