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.
556 lines
14 KiB
556 lines
14 KiB
package service |
|
|
|
import ( |
|
"context" |
|
"fmt" |
|
"time" |
|
|
|
"go-common/app/admin/main/aegis/model" |
|
"go-common/app/admin/main/aegis/model/net" |
|
"go-common/library/ecode" |
|
"go-common/library/log" |
|
) |
|
|
|
//ShowDirection . |
|
func (s *Service) ShowDirection(c context.Context, id int64) (r *net.ShowDirectionResult, err error) { |
|
var ( |
|
d *net.Direction |
|
f *net.Flow |
|
t *net.Transition |
|
) |
|
|
|
if d, err = s.gorm.DirectionByID(c, id); err != nil { |
|
return |
|
} |
|
if f, err = s.gorm.FlowByID(c, d.FlowID); err != nil { |
|
return |
|
} |
|
if t, err = s.gorm.TransitionByID(c, d.TransitionID); err != nil { |
|
return |
|
} |
|
r = &net.ShowDirectionResult{ |
|
Direction: d, |
|
FlowName: f.ChName, |
|
TransitionName: t.ChName, |
|
} |
|
return |
|
} |
|
|
|
//GetDirectionList . |
|
func (s *Service) GetDirectionList(c context.Context, pm *net.ListDirectionParam) (result *net.ListDirectionRes, err error) { |
|
var ( |
|
n *net.Net |
|
flows map[int64]string |
|
trans map[int64]string |
|
unames map[int64]string |
|
fids = []int64{} |
|
tids = []int64{} |
|
uid = []int64{} |
|
) |
|
if n, err = s.gorm.NetByID(c, pm.NetID); err != nil { |
|
return |
|
} |
|
if result, err = s.gorm.DirectionList(c, pm); err != nil { |
|
return |
|
} |
|
if len(result.Result) == 0 { |
|
return |
|
} |
|
for _, item := range result.Result { |
|
fids = append(fids, item.FlowID) |
|
tids = append(tids, item.TransitionID) |
|
uid = append(uid, item.UID) |
|
} |
|
if flows, err = s.gorm.ColumnMapString(c, net.TableFlow, "ch_name", fids, ""); err != nil { |
|
return |
|
} |
|
if trans, err = s.gorm.ColumnMapString(c, net.TableTransition, "ch_name", tids, ""); err != nil { |
|
return |
|
} |
|
if unames, err = s.http.GetUnames(c, uid); err != nil { |
|
log.Error("GetDirectionList s.http.GetUnames error(%v)", err) |
|
err = nil |
|
} |
|
|
|
for _, item := range result.Result { |
|
item.NetName = n.ChName |
|
item.FlowName = flows[item.FlowID] |
|
item.TransitionName = trans[item.TransitionID] |
|
item.UserName = unames[item.UID] |
|
} |
|
return |
|
} |
|
|
|
//SwitchDirection . |
|
func (s *Service) SwitchDirection(c context.Context, id int64, needDisable bool) (err error, msg string) { |
|
var ( |
|
old *net.Direction |
|
action string |
|
canUpdate bool |
|
) |
|
if old, err = s.gorm.DirectionByID(c, id); err != nil { |
|
log.Error("SwitchDirection s.gorm.DirectionByID(%d) error(%v) needDisable(%v)", id, err, needDisable) |
|
return |
|
} |
|
available := old.IsAvailable() |
|
if available == !needDisable { |
|
return |
|
} |
|
|
|
if canUpdate, err = s.beforeUpdate(c, old.NetID); err != nil { |
|
log.Error("SwitchDirection s.beforeUpdate error(%v)", err) |
|
return |
|
} |
|
if !canUpdate { |
|
log.Error("SwitchDirection can't update id(%d) needdisable(%v)", id, needDisable) |
|
return |
|
} |
|
|
|
if needDisable { |
|
old.DisableTime = time.Now() |
|
action = model.LogNetActionDisable |
|
} else { |
|
if err = s.checkDirectionBindAvailable(c, old.FlowID, old.TransitionID); err != nil { |
|
log.Error("SwitchDirection s.checkDirectionBindAvailable(%+v) error(%v) needDisable(%v)", old, err, needDisable) |
|
return |
|
} |
|
if err, msg = s.checkDirConflict(c, old); err != nil { |
|
log.Error("SwitchDirection s.checkDirConflict(%+v) error(%v) needDisable(%v)", old, err, needDisable) |
|
return |
|
} |
|
old.DisableTime = net.Recovered |
|
action = model.LogNetActionAvailable |
|
} |
|
|
|
tx, err := s.gorm.BeginTx(c) |
|
if err != nil { |
|
log.Error("SwitchDirection s.gorm.BeginTx error(%v)", err) |
|
return |
|
} |
|
if err = s.gorm.DisableNet(c, tx, old.NetID); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = s.gorm.UpdateFields(c, tx, net.TableDirection, id, map[string]interface{}{"disable_time": old.DisableTime}); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = tx.Commit().Error; err != nil { |
|
log.Error("SwitchDirection tx.Commit error(%v)", err) |
|
return |
|
} |
|
s.delDirCache(c, old) |
|
|
|
//日志 |
|
oper := &model.NetConfOper{ |
|
OID: old.ID, |
|
Action: action, |
|
UID: old.UID, |
|
NetID: old.NetID, |
|
FlowID: old.FlowID, |
|
TranID: old.TransitionID, |
|
} |
|
s.sendNetConfLog(c, model.LogTypeDirConf, oper) |
|
return |
|
} |
|
|
|
func (s *Service) checkDirectionUnique(c context.Context, netID int64, FlowID int64, transitionID int64, direction int8) (err error, msg string) { |
|
var exist *net.Direction |
|
if exist, err = s.gorm.DirectionByUnique(c, netID, FlowID, transitionID, direction); err != nil { |
|
return |
|
} |
|
if exist != nil { |
|
err = ecode.AegisUniqueAlreadyExist |
|
msg = fmt.Sprintf(ecode.AegisUniqueAlreadyExist.Message(), "有向线", "") |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) checkDirectionBindAvailable(c context.Context, flowID, transitionID int64) (err error) { |
|
var ( |
|
f *net.Flow |
|
t *net.Transition |
|
) |
|
|
|
if flowID > 0 { |
|
if f, err = s.gorm.FlowByID(c, flowID); err != nil { |
|
return |
|
} |
|
if !f.IsAvailable() { |
|
err = ecode.AegisFlowDisabled |
|
return |
|
|
|
} |
|
} |
|
|
|
if transitionID > 0 { |
|
if t, err = s.gorm.TransitionByID(c, transitionID); err != nil { |
|
return |
|
} |
|
if !t.IsAvailable() { |
|
err = ecode.AegisTranDisabled |
|
return |
|
} |
|
} |
|
|
|
return |
|
} |
|
|
|
/** |
|
* 不同order兼容性不同： |
|
* order=顺序，flow<->tran一对一 |
|
* order=todo |
|
*/ |
|
func (s *Service) checkDirConflict(c context.Context, d *net.Direction) (err error, msg string) { |
|
var ( |
|
flowDir, tranDir []*net.Direction |
|
orderDes, conflictDes string |
|
) |
|
if d == nil { |
|
return |
|
} |
|
if flowDir, err = s.gorm.DirectionByFlowID(c, []int64{d.FlowID}, d.Direction); err != nil { |
|
log.Error("checkDirConflict s.gorm.DirectionByFlowID error(%v) direction(%+v)", err, d) |
|
return |
|
} |
|
flowDirLen := len(flowDir) |
|
for k, item := range flowDir { |
|
if item.ID != d.ID { |
|
continue |
|
} |
|
if k < flowDirLen-1 { |
|
flowDir = append(flowDir[:k], flowDir[k+1:]...) |
|
} else { |
|
flowDir = flowDir[:k] |
|
} |
|
flowDirLen-- |
|
} |
|
|
|
if tranDir, err = s.gorm.DirectionByTransitionID(c, []int64{d.TransitionID}, d.Direction, true); err != nil { |
|
log.Error("checkDirConflict s.gorm.DirectionByTransitionID error(%v) direction(%+v)", err, d) |
|
return |
|
} |
|
tranDirLen := len(tranDir) |
|
for k, item := range tranDir { |
|
if item.ID != d.ID { |
|
continue |
|
} |
|
if k < tranDirLen-1 { |
|
tranDir = append(tranDir[:k], tranDir[k+1:]...) |
|
} else { |
|
tranDir = tranDir[:k] |
|
} |
|
tranDirLen-- |
|
} |
|
//无任何已有线 |
|
if flowDirLen == 0 && tranDirLen == 0 { |
|
return |
|
} |
|
|
|
if d.Order == net.DirOrderSequence { |
|
conflictDes = "节点或变迁已被绑定" |
|
} else { |
|
//不支持的顺序报错 |
|
log.Error("checkDirConflict order(%d) is not supported! direction(%+v)", d.Order, d) |
|
err = ecode.RequestErr |
|
return |
|
} |
|
|
|
//顺序兼容性报错 |
|
log.Error("checkDirConflict direction(%+v) not allowed!", d) |
|
err = ecode.AegisDirOrderConflict |
|
orderDes = net.DirOrderDesc[d.Order] |
|
msg = fmt.Sprintf(ecode.AegisDirOrderConflict.Message(), orderDes, conflictDes) |
|
return |
|
} |
|
|
|
//AddDirection . |
|
func (s *Service) AddDirection(c context.Context, uid int64, d *net.DirEditParam) (id int64, err error, msg string) { |
|
var ( |
|
canUpdate bool |
|
) |
|
if canUpdate, err = s.beforeUpdate(c, d.NetID); err != nil { |
|
log.Error("AddDirection s.beforeUpdate error(%v)", err) |
|
return |
|
} |
|
if !canUpdate { |
|
log.Error("AddDirection can't update param(%+v)", d) |
|
return |
|
} |
|
if err, msg = s.checkDirectionUnique(c, d.NetID, d.FlowID, d.TransitionID, d.Direction); err != nil { |
|
return |
|
} |
|
if err = s.checkDirectionBindAvailable(c, d.FlowID, d.TransitionID); err != nil { |
|
return |
|
} |
|
dir := &net.Direction{ |
|
NetID: d.NetID, |
|
FlowID: d.FlowID, |
|
TransitionID: d.TransitionID, |
|
Direction: d.Direction, |
|
Order: d.Order, |
|
Guard: d.Guard, |
|
Output: d.Output, |
|
UID: uid, |
|
} |
|
if err, msg = s.checkDirConflict(c, dir); err != nil { |
|
return |
|
} |
|
|
|
tx, err := s.gorm.BeginTx(c) |
|
if err != nil { |
|
log.Error("AddDirection s.gorm.BeginTx error(%v)", err) |
|
return |
|
} |
|
if err = s.gorm.DisableNet(c, tx, dir.NetID); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = s.gorm.AddItem(c, tx, dir); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = tx.Commit().Error; err != nil { |
|
log.Error("AddDirection tx.Commit error(%v)", err) |
|
return |
|
} |
|
id = d.ID |
|
|
|
//日志 |
|
oper := &model.NetConfOper{ |
|
OID: dir.ID, |
|
Action: model.LogNetActionNew, |
|
UID: dir.UID, |
|
NetID: dir.NetID, |
|
FlowID: dir.FlowID, |
|
TranID: dir.TransitionID, |
|
Diff: []string{ |
|
model.LogFieldTemp(model.LogFieldDirection, net.DirDirectionDesc[dir.Direction], "", false), |
|
model.LogFieldTemp(model.LogFieldOrder, net.DirOrderDesc[dir.Order], "", false), |
|
model.LogFieldTemp(model.LogFieldGuard, dir.Guard, "", false), |
|
model.LogFieldTemp(model.LogFieldOutput, dir.Output, "", false), |
|
}, |
|
} |
|
s.sendNetConfLog(c, model.LogTypeDirConf, oper) |
|
return |
|
} |
|
|
|
//UpdateDirection . |
|
func (s *Service) UpdateDirection(c context.Context, uid int64, d *net.DirEditParam) (err error, msg string) { |
|
var ( |
|
old *net.Direction |
|
canUpdate, checkUnique, orderChanged bool |
|
updates = map[string]interface{}{} |
|
diff = []string{} |
|
) |
|
if old, err = s.gorm.DirectionByID(c, d.ID); err != nil { |
|
log.Error("UpdateDirection s.gorm.DirectionByID(%d) error(%v)", d.ID, err) |
|
return |
|
} |
|
if canUpdate, err = s.beforeUpdate(c, old.NetID); err != nil { |
|
log.Error("UpdateDirection s.beforeUpdate error(%v)", err) |
|
return |
|
} |
|
if !canUpdate { |
|
log.Error("UpdateDirection can't update param(%+v)", d) |
|
return |
|
} |
|
|
|
cp := *old |
|
nw := &cp |
|
if d.FlowID != old.FlowID { |
|
if err = s.checkDirectionBindAvailable(c, d.FlowID, 0); err != nil { |
|
return |
|
} |
|
checkUnique = true |
|
nw.FlowID = d.FlowID |
|
updates["flow_id"] = d.FlowID |
|
} |
|
if d.TransitionID != old.TransitionID { |
|
if err = s.checkDirectionBindAvailable(c, 0, d.TransitionID); err != nil { |
|
return |
|
} |
|
checkUnique = true |
|
nw.TransitionID = d.TransitionID |
|
updates["transition_id"] = d.TransitionID |
|
} |
|
if d.Direction != old.Direction { |
|
checkUnique = true |
|
diff = append(diff, model.LogFieldTemp(model.LogFieldDirection, net.DirDirectionDesc[nw.Direction], net.DirDirectionDesc[old.Direction], true)) |
|
nw.Direction = d.Direction |
|
updates["direction"] = d.Direction |
|
} |
|
if checkUnique { |
|
if err, msg = s.checkDirectionUnique(c, nw.NetID, nw.FlowID, nw.TransitionID, nw.Direction); err != nil { |
|
return |
|
} |
|
} |
|
if d.Order != old.Order { |
|
diff = append(diff, model.LogFieldTemp(model.LogFieldOrder, net.DirOrderDesc[nw.Order], net.DirOrderDesc[old.Order], true)) |
|
nw.Order = d.Order |
|
updates["order"] = d.Order |
|
orderChanged = true |
|
} |
|
//todo-- guard,output |
|
if checkUnique || orderChanged { |
|
if err, msg = s.checkDirConflict(c, nw); err != nil { |
|
return |
|
} |
|
} |
|
if len(updates) <= 0 { |
|
return |
|
} |
|
|
|
tx, err := s.gorm.BeginTx(c) |
|
if err != nil { |
|
log.Error("UpdateDirection s.gorm.BeginTx error(%v)", err) |
|
return |
|
} |
|
if err = s.gorm.DisableNet(c, tx, old.NetID); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = s.gorm.UpdateFields(c, tx, net.TableDirection, old.ID, updates); err != nil { |
|
tx.Rollback() |
|
return |
|
} |
|
if err = tx.Commit().Error; err != nil { |
|
log.Error("UpdateDirection tx.Commit error(%v)", err) |
|
return |
|
} |
|
s.delDirCache(c, old) |
|
|
|
//日志 |
|
oper := &model.NetConfOper{ |
|
OID: nw.ID, |
|
Action: model.LogNetActionUpdate, |
|
UID: nw.UID, |
|
NetID: nw.NetID, |
|
FlowID: nw.FlowID, |
|
TranID: nw.TransitionID, |
|
Diff: diff, |
|
} |
|
s.sendNetConfLog(c, model.LogTypeDirConf, oper) |
|
return |
|
} |
|
|
|
/** |
|
* isDirEnable 检查到下一步的可能性，由order&guard决定 |
|
*/ |
|
func (s *Service) isDirEnable(dir *net.Direction) (enable bool) { |
|
if dir == nil { |
|
return |
|
} |
|
if (dir.Order != net.DirOrderOrSplit && dir.Order != net.DirOrderOrResultSplit) || |
|
(dir.Order == net.DirOrderOrResultSplit && dir.Guard == "") { |
|
enable = true |
|
return |
|
} |
|
|
|
//todo--compute guard expression-v2 |
|
|
|
return |
|
} |
|
|
|
/** |
|
* dispatchDirs 方向线是否可达下一步 |
|
* 加入资源维度：若资源没有任何可达性如何办，这是配置错误--后期加入动态检查 |
|
*/ |
|
func (s *Service) dispatchDirs(dirs []*net.Direction) (enableDir []*net.Direction) { |
|
var ( |
|
enable bool |
|
) |
|
|
|
//对每个有向线, 过滤所有资源, 区分哪些资源可达 |
|
enableDir = []*net.Direction{} |
|
for _, dir := range dirs { |
|
enable = s.isDirEnable(dir) |
|
if enable { |
|
enableDir = append(enableDir, dir) |
|
continue |
|
} |
|
} |
|
|
|
return |
|
} |
|
|
|
/** |
|
* fetchFlowNextEnableDirs 从flow出发，找到下一步变迁 |
|
* 后期加入资源维度：若资源没有任何可达性如何办，这是配置错误--后期加入动态检查 |
|
* 被应用在： |
|
* 1. 新节点查看下一步变迁是否需要创建分发任务; |
|
* 2. 审核提交后，通过flowid找到被触发的是哪个变迁(必须是同一个---后期加入动态检查); |
|
* 3. 通过flowid获取可用变迁的操作项; |
|
* todo -- 任务方存储了transitionid后，可以移除2+3逻辑 |
|
*/ |
|
func (s *Service) fetchFlowNextEnableDirs(c context.Context, flowID int64) (enableDir []*net.Direction, err error) { |
|
var ( |
|
list []*net.Direction |
|
) |
|
|
|
//找到以flow为起点的所有方向线 |
|
if list, err = s.dirByFlow(c, []int64{flowID}, net.DirInput); err != nil { |
|
log.Error("fetchFlowNextEnableDirs s.dirByFlow(%d) error(%v)", flowID, err) |
|
return |
|
} |
|
if len(list) == 0 { //没配置下一步，正常情况 |
|
return |
|
} |
|
|
|
if enableDir = s.dispatchDirs(list); len(enableDir) == 0 { |
|
err = ecode.AegisFlowNoEnableTran |
|
log.Error("fetchFlowNextEnableDirs s.dispatchDirs flowid(%v) no enable transition", flowID) |
|
} |
|
return |
|
} |
|
|
|
/** |
|
* fetchTranNextEnableDirs 从变迁出发，找到下一步flow |
|
* 后期加入资源维度 |
|
*/ |
|
func (s *Service) fetchTranNextEnableDirs(c context.Context, tranID int64) (resultDir *net.Direction, err error) { |
|
var ( |
|
list []*net.Direction |
|
enableDir []*net.Direction |
|
) |
|
if list, err = s.dirByTran(c, []int64{tranID}, net.DirOutput, true); err != nil { |
|
log.Error("fetchTranNextEnableDirs s.dirByTran(%d) error(%v)", tranID, err) |
|
return |
|
} |
|
if len(list) == 0 { //变迁后面没配置flow，为配置错误--后期都需要添加动态检查 |
|
err = ecode.AegisTranNoFlow |
|
log.Error("fetchTranNextEnableDirs s.gorm.DirectionByTransitionID(%d) no flow", tranID) |
|
return |
|
} |
|
|
|
enableDir = s.dispatchDirs(list) |
|
if len(enableDir) == 0 { //变迁后面，没有任何可用flow，为配置错误---后期需添加动态检查 |
|
err = ecode.AegisTranNoFlow |
|
log.Error("fetchTranNextEnableDirs transition(%d) has no enable flow", tranID) |
|
} |
|
|
|
//todo--允许变迁和节点的一对多对应吗？ |
|
resultDir = enableDir[0] |
|
return |
|
} |
|
|
|
func (s *Service) beforeUpdate(c context.Context, netID int64) (ok bool, err error) { |
|
var ( |
|
fr *net.FlowResource |
|
flowID []int64 |
|
) |
|
|
|
if flowID, err = s.flowIDByNet(c, netID); err != nil { |
|
return |
|
} |
|
|
|
if len(flowID) > 0 { |
|
if fr, err = s.gorm.FRByFlow(c, flowID); err != nil { |
|
log.Error("beforeUpdate s.gorm.FRByFlow error(%v) netid(%d)", err, netID) |
|
return |
|
} |
|
} |
|
|
|
ok = fr == nil |
|
return |
|
}
|
|
|