- 1. 编写目的
- 通过对本公司的 tg-flow 软件产品操作过程的描述，使用户可以自主对软件进行使用和操作。
-
- 2. 下载与安装
-
- 2.1 下载安装包
- git clone https://github.com/didi/tg-flow
-
- 2.2 安装和启动
-
- tg-service 的安装和启动：
-
- - 将 tg-service 软件包拷贝到要安装的目录；
- - 进入 tg-service 目录，执行
./build.sh，会生成 output 文件夹；
- - 执行：
cd output 进入 output 目录，执行：nohup ./control.sh start > nohup.log 2>&1 &，启动服务。
-
-
- tg-web 的安装步骤：
-
- - 将 tg-web 软件包拷贝到要安装的目录；
- - 通过命令行方式，进入到 tg-web 所在目录，执行如下命令：
npm install，即可完成 tg-web 部分的安装；
- - 通过命令行方式，进入到 tg-web 所在目录，执行如下命令：
npm start，即可完成 tg-web 的启动；
- - 打开浏览器地址栏，输入 http://localhost:8888/，可以打开登录页，说明安装成功。
-
-
- 3. tg-core 是应用系统使用时需要导入的组件，不需要单独安装。
-
- 3. 登录
-
- 3.1 账号分配
- 初次安装时，系统会默认生成一个管理员账号 admin，密码 admin，可以使用该账号进行后续的登录和分配其他子账号。
-
- 3.2 登录系统
- 第一次打开系统的时候，系统会自动跳转至登录页面，这个时候可以使用管理员账号或者子管理员账号，输入账号、密码、验证码进行登录。
-
-
-
3.3 注销登录
- 进入系统后，可以通过点击系统的右上角“退出登录”按钮注销登录。
-
- 4. 系统管理
- tg-flow 支持同时接入并管理多个系统和应用，每个系统都有独立的系统ID和系统名称，通过系统管理页面可以查看当前接入的所有业务系统。
-
-
-
图4.1 系统管理界面
-
-
- | 属性名称 |
- 描述 |
-
-
- | id |
- 系统 ID |
-
-
- | app_name |
- 系统名称 |
-
-
- | machine_room |
- 系统所部署的集群 |
-
-
- | node_name |
- 系统所在机房 |
-
-
- | operator |
- 系统创建和修改人员 |
-
-
- | create_time |
- 系统创建时间 |
-
-
- | update_time |
- 系统更新时间 |
-
-
- | git_url |
- 系统代码所属的代码仓库地址 |
-
-
-
- 在策略管理 -> 系统管理页面可以进行新增，修改，导出和删除系统等相关操作。
-
-
-
图4.2 新增系统对话框
- 其中系统编号，系统名称，部署机房和节点名称都为必填项，填写完成后点击保存按钮即可完成系统新增。
-
-
-
图4.3 修改系统对话框
- 系统条目右侧的导出按钮并点击确认导出可以导出对应系统所有场景下的流程信息，并作为 zip 压缩包的形式下载到本地并用于对应的系统中。
-
- 5. 场景管理
-
- 5.1 场景列表
- 该列表页展示在当前平台注册的所有的场景信息，系统与场景属于一对多的关系。场景信息所含字段如下图所示。搜索栏中可通过输入“场景名称”或者“系统名称”来获取相应的场景信息。
-
-
-
5.2 场景配置
- 场景列表页存在四个可操作按钮：
-
- - 新增：可手动添加某个系统下的场景信息；
- - 刷新：手动触发获取最新的场景列表；
- - 修改：针对单一场景修改其描述内容，除了场景编号之外其他内容均可修改；
- - 删除：删除某个场景；
-
-
- 6 工作流引擎代码接入
- 6.1 代码框架
- 如下图为一个待接入工作流引擎的应用系统的项目代码，其中logic目录下为业务层代码，接入工作流引擎时，可以在业务层添加相关的接入代码，主要有：
-
- - 工作流引擎初始化代码，类似下图中的workflow_task.go文件；
- - 工作流引擎调度执行代码，类似下图中的dispatcher_genrec.go；
- - 工作流节点对应的算子实现代码，算子有多个，可以放到不同子目录分组存放，类似下图中的module为上述子目录的父目录；
- - 算子注册代码，用于将算子注册到工作流引擎，类似下图中的module_init.go文件.
-
-
-
-
6.2 工作流引擎初始化
- 我们提供了三种工作流引擎初始化的方式：从本地工作流配置文件初始化、从全局配置中心提取配置信息初始化，以及从redis提取配置信息初始化。下面介绍的是第一种，从本地文件初始化，可以在自己的系统初始化模块中调用（其中workflowPath为工作流配置文件根目录）：
-
-
-func LoadWorkflowData(ctx context.Context) {
- moduleObj := module.ModuleObject{}
- workflowEngine, err := wfengine.NewWorkflowEngineFromFile(moduleObj, workflowPath)
- if err != nil {
- tlog.ErrorCount(ctx, "cron_task_workflow_err", fmt.Sprintf("workflow engine init fail, AppId=%v, error=%v", constants.CurrentAppId, err))
- return
- }
-
- global.WorkflowEngine = workflowEngine
- tlog.Handler.Infof(ctx, consts.DLTagCronTask, "workflow engine update successful, appId=%v, workflowEngine=%v", constants.CurrentAppId, global.WorkflowEngine)
-}
-
-
-
- 6.3 工作流引擎调度执行
- 使用工作流引擎调度执行工作流，只需要在调度相关模块中添加下面的代码即可：
-
-
-global.WorkflowEngine.Run(ctx, sc)
-errMap := sc.GetErrorMap()
-errMap.Range(func(key, val interface{}) bool {
- log.Println(fmt.Sprintf("workflowengine error, key=%v, val=%v", key, val))
-)
- return true
-})
-
-
-
- 6.4 工作流算子开发
- 通常一个应用系统中可能会有多个业务场景，每个场景下可能需要配置多个不同的工作流，应用系统对外提供服务时，需要根据实际业务场景或实验分组情况选择使用不同的工作流来执行，工作流则由若干个执行节点组成，每个执行节点对应的是一个算子，在golang语言中这些算子对应的是struct。我们需要在要接入tg-flow的应用系统中定义这些算子，下面为一个名为RecallSampAction的算子定义示例：
-
-
-package recall
-
-import (
- "context"
- "fmt"
- "github.com/didi/tg-flow/tg-core/model"
- "github.com/didi/tg-flow/tg-core/wfengine"
- "time"
-)
-
-type RecallSampAction struct {
- wfengine.ModelBase
-}
-
-func (r RecallSampAction) DoAction(ctx context.Context, sc *model.StrategyContext) interface{} {
- //此处添加相关业务逻辑
- return nil
-}
-
-func (r RecallSampAction) OnTimeout(context.Context, *model.StrategyContext){
- //此处添加超时处理逻辑
-}
-
-
-
- 6.5 工作流算子注册
- 为了将开发好的算子注册给工作流引擎，需要进行算子注册，如下图中的case语句部分，即为工作流算子注册函数，我们只需要自己实现一个下面的函数，并在函数中的case语句中添加自己的实际开发的算子相关注册代码即可：
-
-
-func (moduleObj ModuleObject) NewObj(moduleName string, vMap map[string]string) wfengine.IModelBase {
- switch moduleName {
- case “data.DataPrepareSamp1”: return &data.DataPrepareSamp1{}
- case “recall.RecallSamp1”: return &recall.RecallSamp1{}
- case “recall.RecallSamp2”: return &recall.RecallSamp2{}
- }
- return nil
-}
-
-
-
- 6.6 代码仓库注册
- 在应用系统中完成上述代码编写后，可以将代码提交到git代码仓库，并将git仓库地址拷贝出来，然后进入系统管理页面，如下图所示，点击“修改”按钮，将该代码仓库的地址填写到弹出层中的“git仓库”文本框，如下图红色矩形框所示，然后点击“保存“按钮即可。
-
-
-
7 工作流管理
- 7.1 工作流列表
- 点击左侧的“流程管理”目录，可以打开流程列表的管理页面，如下图所示。
-
-
-
-
- 选择列表上方的“系统”下拉框，点击查询，可以查到对应的应用系统已经添加的所有流程。
- 选择“场景名称”，点击查询，可以列出指定系统及场景下的所有工作流。
- 点击“新建”按钮，可以打开工作流编辑页面，开始一个新工作流的创建，下节会详细介绍新工作流的编辑功能。
- 工作流列表中有：修改、导出、复制、导入、删除等5个按钮。
- 点击“修改“按钮，可以打开工作流编辑页，后面的工作流编辑部分会详细介绍。
- 点击“导出”按钮，可以将当前工作流配置信息导出为一个json文件，点击确认后可以下载到本地，后面可以拷贝到待接入的应用系统工程目录下，应用系统在启动时，会初始化工作流引擎，工作流引擎初始化时可以通过本地方式加载上述配置。注意：此处是对单个工作流配置进行导出，如果要将某个系统下所有工作流打包导出，则需要进入到系统管理页操作。
- 点击“复制”按钮，可以新建一个同样的工作流配置，然后后面就可以在这个工作流上进行相关修改并保存。
- 点击“导入”按钮，会弹出一个对话框，这时可以将在外部编写的工作流配置信息复制并粘贴到这个对话框中的“导入内容（Json）”文本框中，点击“确认”按钮，即可完成工作流配置信息的导入，注意：工作流节点中的action_id的取值格式为：action-{场景ID}-{工作流ID}，导入工作流时如果相关actionId中的场景ID和工作流ID与当前要导入的场景ID和工作流ID不一致，系统会自动帮忙将actionId中的后面两部分的值调整为当前的场景ID和工作流ID。
-
-
-
7.2 工作流编辑
- 打开工作流编辑页，系统会自动读取前面配置好的git仓库中的代码模块，获取代码中的算子列表，并在工作流编辑页面的左侧算子列表中展示，供后续的工作流编辑用。(注意：工作流中的条件节点已经在工作流引擎中做了内置的模块实现，所以无需单独在应用层再做实现，当然如果有需要，也可以自定义条件节点后，注册到工作流引擎)。如下图，为工作流编辑页面。工作流编辑页面包括三个区域：左侧为工作流节点列表，中间为流程构建区域，右侧为工作流及节点属性编辑区域。
-
-
-
其中：
- 在后台管理系统中打开工作流编辑页面时，系统会从后台管理系统中配置的代码仓库中拉取代码模块名称，并以组件的方式显示在左侧组件栏。左侧的组件有两种类型：一是应用层节点，用户打开工作流图编辑界面后，会自动从对应的应用代码的git仓库中拉取对应的算子列表，展示在左侧的节点列表中。二是工作流引擎内置的条件节点和缺省节点（当git仓库未完成代码提交时，左侧无法拉取到代码模块名称，这时可以手动添加缺省节点，然后给缺省节点重新命名）。
- 从左侧组件栏中依次拖拽节点、添加节点之间的连线，可以逐步构建出类似下图中的主流程图和子流程图。
- 针对图中的每个节点，可以依次点击节点，然后在右侧的属性编辑区域编辑该节点的属性，如：节点名称、节点参数、节点描述信息等，完成所有节点属性的编辑后点击“保存”按钮，相关的工作流配置信息便会以json字符串格式存入数据库。
-
-
-
-
7.3 工作流配置发布
- 上一节中已存入数据库的工作流配置信息并不能自动发布到在线应用系统，需要在系统管理页面中点击“提交”按钮，将上述配置发布到一个具有存储全局配置功能的存储服务中，一般可基于zookeeper或redis开发，以供应用系统从配置中心读取上述配置信息。相应地，我们在工作流引擎也提供了从redis、zookeeper及本地文件中加载工作流配置的方法，前面工作流引擎接入部分已有介绍不作详述。
-
-
- 7.4 工作流导出与导入
- 然而，由于部分应用系统出于安全或者降低外部依赖的角度考虑，并不希望通过与后台管理系统进行联动的方式，动态更新工作流配置信息，因此我们在系统管理页面提供了工作流配置信息导出功能，用户可以直接将工作流配置信息导出为zip包，然后将上述zip包解压后直接添加到其应用系统的项目的本地目录下，应用系统在初始化工作流引擎时，可以直接使用工作流引擎提供的相关方法从本地目录加载工作流配置。
-
-
-
-
\ No newline at end of file
diff --git a/tg-core/wfengine/condition.go b/wfengine/condition.go
similarity index 100%
rename from tg-core/wfengine/condition.go
rename to wfengine/condition.go
diff --git a/tg-core/wfengine/engine.go b/wfengine/engine.go
similarity index 67%
rename from tg-core/wfengine/engine.go
rename to wfengine/engine.go
index 2d84e3acb0c60b30b0f5c01cca3b8131194b6f1e..522a1f7d4a0dd64bd5a010320314fdda36781d94 100644
--- a/tg-core/wfengine/engine.go
+++ b/wfengine/engine.go
@@ -1,4 +1,5 @@
-/**
+/*
+*
Description : workflow engine v3.1
Author : dayunzhangyunfeng@didiglobal.com
Date : 2021-05-14
@@ -9,9 +10,9 @@ import (
"context"
"errors"
"fmt"
- "github.com/didi/tg-flow/tg-core/common/tlog"
- "github.com/didi/tg-flow/tg-core/consts"
- "github.com/didi/tg-flow/tg-core/model"
+ "github.com/didi/tg-flow/common/tlog"
+ "github.com/didi/tg-flow/consts"
+ "github.com/didi/tg-flow/model"
"strings"
"sync"
"time"
@@ -25,37 +26,63 @@ const (
type WorkflowEngine struct {
sceneModuleMap map[int64]*SceneModule `json:"scene_module_map"`
workflowMap map[int64]*Workflow `json:"workflows"`
- modelBaseMap map[string]IModelBase `json:"modules"`
+ modelBaseMap map[string]IModelBase `json:"modules"`
updateTime string `json:"update_time"`
condExecutors *CondExecutors `json:"cond_executors"`
+ FlowSelectors map[int]FlowSelector `json:"flow_selectors"`
}
-func NewWorkflowEngine(moduleObj ModuleObjBase, appId int64) (*WorkflowEngine, error) {
- //更新系统下全部场景的节点对象
- smMap, err := LoadSceneModuleMap(appId)
- if err != nil {
- return nil, err
- }
+func (w *WorkflowEngine) SetCustomFlowSelector(flowSelector FlowSelector) {
+ w.FlowSelectors[consts.FLOW_BY_CUSTOM] = flowSelector
+}
- //load workflow
- wfMap, err := LoadWorkflow(appId, smMap)
- if err != nil {
- return nil, err
+func (w *WorkflowEngine) GetWorkflow(workflowId int64) (*Workflow, bool) {
+ workflow, ok := w.workflowMap[workflowId]
+ return workflow, ok
+}
+
+func (w *WorkflowEngine) SelectWorkflow(ctx context.Context, sc *model.StrategyContext) (*Workflow, error) {
+ sceneModule, okE := w.sceneModuleMap[sc.SceneId]
+ if !okE {
+ return nil, fmt.Errorf("no sceneModule found for scene_id:%v", sc.SceneId)
}
- //TODO ZYF err
- resetWorkflows(wfMap)
+ var flowId int64
+ var groupName string
+ var err error
+ if sc.FlowId > 0 { //测试
+ flowId = sc.FlowId
+ } else {
+ //根据该场景设置的分流方式，获取workflowId
+ fs, ok := w.FlowSelectors[sceneModule.FlowType]
+ if !ok {
+ return nil, fmt.Errorf("no flow selector found for your flow type:%v", sceneModule.FlowType)
+ }
- mbMap, err := createModelMap(moduleObj, wfMap)
- if err != nil {
- return nil, err
+ flowId, groupName, err = fs.SelectWorkflowId(sc, sceneModule)
+ // use default workflow if no flowId found
+ if err != nil || flowId <= 0 {
+ flowId = sceneModule.DefaultWorkflowId
+ tlog.Handler.ErrorCount(ctx, "select_workflow_err", fmt.Sprintf("select workflow error:%v, use default workflow:%v", err, flowId))
+ err = nil
+ }
+
+ sc.FlowId = flowId
+ }
+ //根据分流选出的workflow_id，取得对应的实验策略配置
+ flow, okW := w.GetWorkflow(sc.FlowId)
+ if !okW {
+ return nil, fmt.Errorf("no workflow found, flowId=%v||groupId=%v||err=%v", sc.FlowId, groupName, err)
}
- workfowEngine := newWorkflowEngine(smMap, wfMap, mbMap)
- return workfowEngine, nil
+ sc.Set("flow", flow)
+ sc.Set("groupId", groupName)
+ sc.Set("scene", sceneModule)
+ return flow, nil
}
func resetWorkflows(wfMap map[int64]*Workflow) {
+ //fmt.Println("wfMap.len=", len(wfMap))
if len(wfMap) == 0 {
return
}
@@ -63,7 +90,7 @@ func resetWorkflows(wfMap map[int64]*Workflow) {
for _, workflow := range wfMap {
err := resetWorkflow(wfMap, workflow)
if err != nil {
- tlog.ErrorCount(context.TODO(), "resetWorkflow err", fmt.Sprintf("workflow=%+v", workflow))
+ tlog.Handler.ErrorCount(context.TODO(), "resetWorkflow err", fmt.Sprintf("workflow=%+v", workflow))
continue
}
}
@@ -73,28 +100,29 @@ func resetWorkflow(wfMap map[int64]*Workflow, workflow *Workflow) error {
if workflow == nil {
return errors.New("workflow is nil:%v")
}
-
+ //fmt.Println("开始for循环处理flowAction")
for {
flowActionMap := make(map[string]*Action)
+ //fmt.Println("workflow.FlowCharts.ActionMap.len===>", len(workflow.FlowCharts.ActionMap), workflow.FlowCharts.HashCondition)
for _, action := range workflow.FlowCharts.ActionMap {
if action.ActionType == ActionTypeFlow {
if action.RefWorkflowId <= 0 {
- tlog.ErrorCount(context.TODO(), "resetWorkflow_err", fmt.Sprintf("ref_workflow_id must >0, flowAction.Action=%+v", action))
- continue
+ tlog.Handler.ErrorCount(context.TODO(), "resetWorkflow_err", fmt.Sprintf("ref_workflow_id must >0, flowAction.Action=%+v", action))
+ continue
}
flowActionMap[action.ActionId] = action
}
}
if len(flowActionMap) < 1 {
- return nil
+ break
}
-
+ //fmt.Println()
//对所有flowAction
for _, flowAction := range flowActionMap {
//get workflowid
refWf := wfMap[flowAction.RefWorkflowId]
if refWf == nil {
- tlog.ErrorCount(context.TODO(), "resetWorkflow_err", fmt.Sprintf("reference workflow not exist, flowAction.Action=%+v", flowAction))
+ tlog.Handler.ErrorCount(context.TODO(), "resetWorkflow_err", fmt.Sprintf("reference workflow not exist, flowAction.Action=%+v", flowAction))
continue
}
@@ -108,13 +136,13 @@ func resetWorkflow(wfMap map[int64]*Workflow, workflow *Workflow) error {
firstActionId := refWf.FlowCharts.FirstActionId
if flowAction.ActionId == workflow.FlowCharts.FirstActionId {
workflow.FlowCharts.FirstActionId = firstActionId
- }else {
+ } else {
for _, prevId := range flowAction.PrevActionIds {
strNextActionIds := strings.Join(workflow.FlowCharts.ActionMap[prevId].NextActionIds, ",")
strNewNextActionIds := ""
- if flowAction.Timeout >0 {
+ if flowAction.Timeout > 0 {
strNewNextActionIds = strNextActionIds + "," + firstActionId
- }else {
+ } else {
strNewNextActionIds = strings.ReplaceAll(strNextActionIds, flowAction.ActionId, firstActionId)
}
workflow.FlowCharts.ActionMap[prevId].NextActionIds = strings.Split(strNewNextActionIds, ",")
@@ -128,7 +156,7 @@ func resetWorkflow(wfMap map[int64]*Workflow, workflow *Workflow) error {
lastActionId := refWf.FlowCharts.LastActionId
if flowAction.ActionId == workflow.FlowCharts.LastActionId {
workflow.FlowCharts.LastActionId = lastActionId
- }else {
+ } else {
for _, nextId := range flowAction.NextActionIds {
strPrevActionIds := strings.Join(workflow.FlowCharts.ActionMap[nextId].PrevActionIds, ",")
strNewPrevActionIds := ""
@@ -146,72 +174,48 @@ func resetWorkflow(wfMap map[int64]*Workflow, workflow *Workflow) error {
//删除
if flowAction.Timeout > 0 {
flowAction.ActionType = ActionTypeTimeout
- }else{
+ } else {
delete(workflow.FlowCharts.ActionMap, flowAction.ActionId)
}
}
}
+ //fmt.Println("before workflow.FlowCharts.HashCondition=>", workflow.FlowCharts.HashCondition)
for _, action := range workflow.FlowCharts.ActionMap {
+ //fmt.Println("actionId=", action.ActionId, "action.ActionType=", action.ActionType)
if action.ActionType == ActionTypeCond {
workflow.FlowCharts.HashCondition = true
break
}
- }
- return nil
-}
-func NewWorkflowEngineFromFile(moduleObj ModuleObjBase, configPath string) (*WorkflowEngine, error) {
- smMap, err := LoadSceneModuleMapFromFile(configPath + "/scene.json")
- if err != nil {
- return nil, err
}
-
- //更新workflow
- wfMap, err := LoadWorkflowFromFile(configPath, smMap)
- if err != nil {
- return nil, err
- }
-
- resetWorkflows(wfMap)
-
- mbMap, err := createModelMap(moduleObj, wfMap)
- if err != nil {
- return nil, err
- }
-
- workfowEngine := newWorkflowEngine(smMap, wfMap, mbMap)
- return workfowEngine, nil
+ //fmt.Println("after workflow.FlowCharts.HashCondition, workflowId=",workflow.Id, "hasCondition=",workflow.FlowCharts.HashCondition)
+ return nil
}
-func NewWorkflowEngineFromApollo(moduleObj ModuleObjBase, namespace, configName string) (*WorkflowEngine, error) {
- smMap, wfMap, err := LoadSceneModuleWorkflowFromApollo(namespace, configName)
- if err != nil {
- return nil, err
- }
- resetWorkflows(wfMap)
+func NewWorkflowEngine(sceneModuleMap map[int64]*SceneModule, workflowMap map[int64]*Workflow, version string, moduleObj ModuleObjBase) (*WorkflowEngine, error) {
+ //TODO ZYF err
+ resetWorkflows(workflowMap)
- mbMap, err := createModelMap(moduleObj, wfMap)
+ modelBaseMap, err := createModelMap(moduleObj, workflowMap)
if err != nil {
return nil, err
}
-
- workfowEngine := newWorkflowEngine(smMap, wfMap, mbMap)
- return workfowEngine, nil
-}
-
-func newWorkflowEngine(sceneModuleMap map[int64]*SceneModule, workflowMap map[int64]*Workflow, modelBaseMap map[string]IModelBase) *WorkflowEngine {
- ut := fmt.Sprintf("%v", time.Now().Format("2006-01-02 15:04:05"))
+ //ut := fmt.Sprintf("%v", time.Now().Format("2006-01-02 15:04:05"))
cExecutors := GetCondExecutors()
+ flowSelectors := make(map[int]FlowSelector)
+ flowSelectors[consts.FLOW_BY_ONLINE_RANDOM] = &RandomSelector{}
+ flowSelectors[consts.FLOW_BY_APOLLO] = &GroupSelector{}
wfe := &WorkflowEngine{
sceneModuleMap: sceneModuleMap,
workflowMap: workflowMap,
modelBaseMap: modelBaseMap,
- updateTime: ut,
+ updateTime: version,
condExecutors: cExecutors,
+ FlowSelectors: flowSelectors,
}
- return wfe
+ return wfe, nil
}
func createModelMap(moduleObj ModuleObjBase, wfMap map[int64]*Workflow) (map[string]IModelBase, error) {
@@ -223,7 +227,7 @@ func createModelMap(moduleObj ModuleObjBase, wfMap map[int64]*Workflow) (map[str
modelBaseMap := make(map[string]IModelBase)
for _, wf := range wfMap {
if wf.FlowCharts == nil {
- tlog.ErrorCount(ctx, "create_modelbase_err", fmt.Sprintf("flow_charts is nil, workflow=%v", wf))
+ tlog.Handler.ErrorCount(ctx, "create_modelbase_err", fmt.Sprintf("flow_charts is nil, workflow=%v", wf))
continue
}
@@ -234,7 +238,7 @@ func createModelMap(moduleObj ModuleObjBase, wfMap map[int64]*Workflow) (map[str
mb, err := createModelBase(moduleObj, action)
if mb == nil || err != nil {
- tlog.ErrorCount(ctx, "create_modelbase_err", fmt.Sprintf("workflow=%v, action=%v, mb=%v, err=%v", wf, action, mb, err))
+ tlog.Handler.ErrorCount(ctx, "create_modelbase_err", fmt.Sprintf("workflow=%v, action=%v, mb=%v, err=%v", wf, action, mb, err))
continue
}
modelBaseMap[actionId] = mb
@@ -250,24 +254,31 @@ func createModelBase(moduleObj ModuleObjBase, action *Action) (IModelBase, error
return nil, fmt.Errorf("action or moduleObj empty, action:%v,moduleObj:%v", action, moduleObj)
}
+ mb := moduleObj.NewObj(action.ActionName)
+ if mb == nil {
+ return mb, fmt.Errorf("create ModelBase instance error, action:%v", action)
+ }
+
+ mb.SetName(action.ActionName)
+
vMap := make(map[string]string)
if len(action.Params) > 0 {
for _, param := range action.Params {
vMap[param.Name] = param.Value
}
}
- vMap["Name"] = action.ActionName
- mb := moduleObj.NewObj(action.ActionName, vMap)
- if mb == nil {
- return mb, fmt.Errorf("create ModelBase instance error, action:%v, vMap:%v", action, vMap)
+ err := reflectModuleField(mb, vMap)
+ if err != nil {
+ tlog.Handler.ErrorCount(context.TODO(), "createModelBase_err", fmt.Sprintf("set module field fail, actionName:%v, vMap:%v, error:%v", action.ActionName, vMap, err))
}
return mb, nil
}
-func (w *WorkflowEngine) GetUpdateTime() string {
+func (w *WorkflowEngine) GetVersion() string {
return w.updateTime
}
+
func (w *WorkflowEngine) RegisterCondExecutor(conditionName string, executor interface{}) {
w.condExecutors.RegisterCondExecutor(conditionName, executor)
}
@@ -279,10 +290,10 @@ func (w *WorkflowEngine) Run(ctx context.Context, sc *model.StrategyContext) {
sc.SetError(action0, err)
}
}()
- //选择一个workflow策略
- flow, err := w.selectWorkFlow(ctx, sc)
+
+ flow, err := w.SelectWorkflow(ctx, sc)
if err != nil {
- sc.SetError(action0, fmt.Errorf("flow:%v, err:%v", flow, err))
+ sc.SetError(action0, fmt.Errorf("no workflow found, flowId=%v, err=%v", sc.FlowId, err))
return
}
@@ -292,8 +303,7 @@ func (w *WorkflowEngine) Run(ctx context.Context, sc *model.StrategyContext) {
return
}
- wgMap,tsMap := flowChart.CreateWaitMap()
-
+ wgMap, tsMap := flowChart.CreateWaitMap()
waitedMap := &sync.Map{}
wgn := &sync.WaitGroup{}
wgn.Add(1)
@@ -303,51 +313,14 @@ func (w *WorkflowEngine) Run(ctx context.Context, sc *model.StrategyContext) {
wgn.Wait()
}
-func (w *WorkflowEngine) selectWorkFlow(ctx context.Context, sc *model.StrategyContext) (*Workflow, error) {
- sceneModule, okE := w.sceneModuleMap[sc.SceneId]
- if !okE {
- return nil, fmt.Errorf("no sceneModule found for scene_id:%v", sc.SceneId)
- }
+/*
+*
- var flowId int64
- var slotId int
- var groupName string
- var err error
- if sc.FlowId > 0 { //测试
- flowId = sc.FlowId
- } else {
- //根据该场景设置的分流方式，获取workflowId
- if sceneModule.FlowType == consts.FLOW_BY_ONLINE_RANDOM {
- flowId, slotId = FlowByOnlineRandom(sc, sceneModule)
- } else if sceneModule.FlowType == consts.FLOW_BY_APOLLO {
- flowId, groupName, err = FlowByApollo(sc, sceneModule)
- }
-
- // 如果没有找到 flowId，采用默认的 Workflow
- if err != nil || flowId == 0 {
- tlog.ErrorCount(ctx, "select_workflow_err", fmt.Sprintf("select workflow by apollo error, err=%v, use default workflow", err))
- flowId = sceneModule.DefaultWorkflowId
- err = nil
- }
-
- sc.FlowId = flowId
- }
- //根据分流选出的workflow_id，取得对应的实验策略配置
- flow, okW := w.workflowMap[flowId]
- if !okW {
- err := fmt.Errorf("no workflow found, slotId=%v||flowId=%v||groupId=%v||err=%v", slotId, flowId, groupName, err)
- return nil, err
- }
- sc.Set("flow", flow)
- sc.Set("groupId", slotId)
- sc.Set("scene", sceneModule)
- return flow, err
-}
-
-/**
新版本（v1.2.11）后因为有动态条件节点，原流程执行模式有较大变化，不再包含预处理，需实时判断节点走向和执行、跳过节点等操作。
-**/
-func (w *WorkflowEngine) doExecuteModule(ctx context.Context, sc *model.StrategyContext, flowChart *WorkflowChart, skipedActionIdPairs *sync.Map, wgMap map[string]*sync.WaitGroup, tsMap map[string]*TimeWaiter, waitedMap *sync.Map, actionId string, wgn *sync.WaitGroup) {
+
+*
+*/
+func (w *WorkflowEngine) doExecuteModule(ctx context.Context, sc *model.StrategyContext, flowChart *WorkflowChart, skipedActionIdPairs *sync.Map, wgMap map[string]*sync.WaitGroup, tsMap map[string]*TimeWaiter, waitedMap *sync.Map, actionId string, wgn *sync.WaitGroup) {
action, ok := flowChart.ActionMap[actionId]
if !ok {
return
@@ -370,7 +343,7 @@ func (w *WorkflowEngine) doExecuteModule(ctx context.Context, sc *model.Strategy
if timeoutActionId == "" {
wgMap[action.ActionId].Wait()
- }else {
+ } else {
go func() {
wgMap[action.ActionId].Wait()
if tsMap[timeoutActionId] != nil {
@@ -382,7 +355,7 @@ func (w *WorkflowEngine) doExecuteModule(ctx context.Context, sc *model.Strategy
if mb, _ := w.modelBaseMap[timeoutActionId]; mb != nil {
if flowChart.ActionMap[timeoutActionId].TimeoutAsync {
go mb.OnTimeout(ctx, sc)
- }else{
+ } else {
mb.OnTimeout(ctx, sc)
}
}
@@ -412,13 +385,13 @@ func (w *WorkflowEngine) doExecuteModule(ctx context.Context, sc *model.Strategy
}
-func (w *WorkflowEngine) skipBranch(flowChart *WorkflowChart, wgMap map[string]*sync.WaitGroup, skipedActionIdPairs *sync.Map, toExcludeActionId string, prevAction *Action, action *Action){
+func (w *WorkflowEngine) skipBranch(flowChart *WorkflowChart, wgMap map[string]*sync.WaitGroup, skipedActionIdPairs *sync.Map, toExcludeActionId string, prevAction *Action, action *Action) {
if prevAction == nil || action == nil {
return
}
- if len(action.PrevActionIds)>1 {
- if _, ok := skipedActionIdPairs.LoadOrStore(prevAction.ActionId + "_" + action.ActionId, ""); !ok {
+ if len(action.PrevActionIds) > 1 {
+ if _, ok := skipedActionIdPairs.LoadOrStore(prevAction.ActionId+"_"+action.ActionId, ""); !ok {
wgMap[action.ActionId].Done()
}
}
@@ -429,46 +402,51 @@ func (w *WorkflowEngine) skipBranch(flowChart *WorkflowChart, wgMap map[string]*
return
}
- for _, nextActionId := range action.NextActionIds {
+ nextCount := len(action.NextActionIds)
+ for i := nextCount - 1; i >= 0; i-- {
+ nextActionId := action.NextActionIds[i]
nextAction := flowChart.ActionMap[nextActionId]
w.skipBranch(flowChart, wgMap, skipedActionIdPairs, toExcludeActionId, action, nextAction)
}
}
-func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyContext, flowChart *WorkflowChart, action *Action, skipedActionIdPairs *sync.Map, wgMap map[string]*sync.WaitGroup, tsMap map[string]*TimeWaiter, wgn *sync.WaitGroup) string {
+func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyContext, flowChart *WorkflowChart, action *Action, skipedActionIdPairs *sync.Map, wgMap map[string]*sync.WaitGroup, tsMap map[string]*TimeWaiter, wgn *sync.WaitGroup) string {
toExeActionId := ""
defer func() {
if err := recover(); err != nil {
- tlog.ErrorCount(ctx, "executeModule_err", fmt.Sprintf("actionId=%+v,err=%+v", action.ActionId, err))
+ tlog.Handler.ErrorCount(ctx, "executeModule_err", fmt.Sprintf("actionId=%+v,err=%+v", action.ActionId, err))
sc.SetError(action.ActionId, fmt.Errorf("%v", err))
sc.Skip(ErrNoUnknown, fmt.Sprintf("executeModule_err, actionId=%+v, err=%+v", action.ActionId, err))
}
- for _, nextActionId := range action.NextActionIds {
+ nextCount := len(action.NextActionIds)
+ for i := nextCount - 1; i >= 0; i-- {
+ nextActionId := action.NextActionIds[i]
nextAction, ok := flowChart.ActionMap[nextActionId]
if !ok {
continue
}
if action.ActionType == ActionTypeCond {
- //condition
if nextActionId != toExeActionId {
w.skipBranch(flowChart, wgMap, skipedActionIdPairs, toExeActionId, action, nextAction)
}
- }else{
- //task
- if len(nextAction.PrevActionIds) > 1 {
+ }
+
+ if len(nextAction.PrevActionIds) > 1 {
+ if _, ok := skipedActionIdPairs.LoadOrStore(action.ActionId+"_"+nextActionId, ""); !ok {
wgMap[nextActionId].Done()
}
}
+
}
if action.ActionId == flowChart.LastActionId {
wgn.Done()
}
- //fmt.Println(fmt.Sprintf("====>Finish time=%v ,actionId=%v, actionName=%v", time.Now().Format("2006-01-02 15:04:05.000"), action.ActionId, action.ActionName))
+ //fmt.Println(fmt.Sprintf("====>Finish time=%v ,actionId=%v, actionName=%v, action=%+v", time.Now().Format("2006-01-02 15:04:05.000"), action.ActionId, action.ActionName,action))
}()
if action == nil {
@@ -479,7 +457,7 @@ func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyCo
var err error
toExeActionId, err = action.executeCond(sc.ContextMap)
if err != nil {
- tlog.ErrorCount(ctx, "executeCond_err", fmt.Sprintf("execute condition error, actionId=%+v, toExeActionId=%+v, err=%+v", action.ActionId, toExeActionId, err))
+ tlog.Handler.ErrorCount(ctx, "executeCond_err", fmt.Sprintf("execute condition error, actionId=%+v, toExeActionId=%+v, err=%+v", action.ActionId, toExeActionId, err))
sc.SetError(action.ActionId, fmt.Errorf("action.executeCond error, ActionName:%v, err:%v", action.ActionName, err))
}
return toExeActionId
@@ -488,10 +466,10 @@ func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyCo
if moduleBase, ok := w.modelBaseMap[action.ActionId]; ok {
if !sc.IsSkip() {
startTime := time.Now().UnixNano() / 1e6
- var result interface{}
- if action.Timeout>0 && action.ActionType != ActionTypeTimeout {
+ sc.TC.StartSectionCount(action.ActionName)
+ if action.Timeout > 0 && action.ActionType != ActionTypeTimeout {
go func() {
- result = moduleBase.DoAction(ctx, sc)
+ moduleBase.DoAction(ctx, sc)
if tsMap[action.ActionId] != nil {
tsMap[action.ActionId].Done()
}
@@ -500,10 +478,10 @@ func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyCo
sc.AddTimeoutAction(action.ActionName)
if action.TimeoutAsync {
go moduleBase.OnTimeout(ctx, sc)
- }else{
+ } else {
moduleBase.OnTimeout(ctx, sc)
}
- }else{
+ } else {
if action.TimeoutDynamic {
for _, nextActionId := range action.NextActionIds {
if ts, ok := tsMap[nextActionId]; ok {
@@ -513,12 +491,11 @@ func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyCo
}
}
}
- }else{
- result = moduleBase.DoAction(ctx, sc)
+ } else {
+ moduleBase.DoAction(ctx, sc)
}
- endTime := time.Now().UnixNano() / 1e6
- sc.SetModuleResult(action.ActionId, action.ActionName, endTime-startTime, result)
+ sc.TC.StopSectionCount(action.ActionName)
}
} else {
sc.SetError(action.ActionId, fmt.Errorf("module not found in map, moduleMap:%v, moduleName:%v", w.modelBaseMap, action.ActionName))
@@ -527,4 +504,3 @@ func (w *WorkflowEngine) executeModule(ctx context.Context, sc *model.StrategyCo
return ""
}
-
diff --git a/tg-core/wfengine/workflow_apollo_dao.go b/wfengine/engine_from_apollo.go
similarity index 39%
rename from tg-core/wfengine/workflow_apollo_dao.go
rename to wfengine/engine_from_apollo.go
index b1374ceace17a36fd5e1515bb61f0c7951ccc6d1..c37c8c091d7c30bf865d5ce9a2347faa1f00ebad 100644
--- a/tg-core/wfengine/workflow_apollo_dao.go
+++ b/wfengine/engine_from_apollo.go
@@ -1,7 +1,7 @@
/**
- Description : loader of workflow config info from apollo config
- Author : dayunzhangyunfeng@didiglobal.com
- Date : 2023-01-09 14:30
+Description : loader of workflow config info from apollo config
+Author : dayunzhangyunfeng@didiglobal.com
+Date : 2023-01-09 14:30
*/
package wfengine
@@ -10,52 +10,65 @@ import (
"context"
"encoding/json"
"fmt"
- "github.com/didi/tg-flow/tg-core/common/tlog"
- "github.com/didi/tg-flow/tg-core/wfengine/apollo"
+ "github.com/didi/tg-flow/common/tlog"
"strconv"
"strings"
)
const (
- dftRange = "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99"
- sceneKey = "scene"
- )
-
-func LoadSceneModuleWorkflowFromApollo(namespace, configName string) (map[int64]*SceneModule, map[int64]*Workflow, error) {
- apolloConfig, err := apollo.NewApolloConfig(namespace, configName)
- if err != nil || apolloConfig == nil {
- return nil, nil, fmt.Errorf("init apolloConfig fail, namespace=%v, configName=%v, err=%v", namespace, configName, err)
+ dftRange = "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99"
+ sceneKey = "scene"
+ versionKey = "version"
+)
+
+//func GetLatestVersionFromApollo(namespace, configName string) (string, error) {
+// apolloConfig, err := apollo.NewApolloConfig(namespace, configName)
+// if err != nil || apolloConfig == nil {
+// return "", fmt.Errorf("NewApolloConfig error, namespace=%v, configName=%v, err=%v", namespace, configName, err)
+// }
+//
+// configParams := apolloConfig.GetConfigs()
+// if configParams == nil || configParams[versionKey] == "" {
+// return "", fmt.Errorf("GetConfigs fail, configParam is nil or version empty, namespace=%v, configName=%v, configParams.lenth=%v", namespace, configName, len(configParams))
+// }
+//
+// return configParams[versionKey], nil
+//}
+
+func NewWorkflowEngineFromConfig(moduleObj ModuleObjBase, configParams map[string]string) (*WorkflowEngine, error) {
+ smMap, wfMap, version, err := loadSceneModuleWorkflowFromApollo(configParams)
+ if err != nil {
+ return nil, err
}
- configParams := apolloConfig.GetConfigs()
- if configParams == nil || configParams[sceneKey] == "" {
- return nil, nil, fmt.Errorf("apollo GetConfigs fail, some keys not found namespace=%v, configName=%v, configParams=%v", namespace, configName, configParams)
+ return NewWorkflowEngine(smMap, wfMap, version, moduleObj)
+}
+
+func loadSceneModuleWorkflowFromApollo(configParams map[string]string) (map[int64]*SceneModule, map[int64]*Workflow, string, error) {
+ if configParams == nil {
+ return nil, nil, "", fmt.Errorf("GetConfigs error, configParams is nil")
}
- sceneModuleString := ""
- for key, val := range configParams {
- if key == sceneKey {
- sceneModuleString = val
- break
- }
+ if configParams[sceneKey] == "" {
+ return nil, nil, "", fmt.Errorf("configParams error, sceneKey[%v] is empty", sceneKey)
}
- smMap, err := loadSceneModuleString(sceneModuleString)
+ smMap, err := loadSceneModuleString(configParams[sceneKey])
if err != nil {
- return nil, nil, err
+ return nil, nil, "", err
}
//更新workflow
wfMap, err := LoadWorkflowFromApollo(configParams)
if err != nil {
- return nil, nil, err
+ return nil, nil, "", err
}
- return smMap, wfMap, nil
+ return smMap, wfMap, configParams[versionKey], nil
}
func loadSceneModuleString(sceneModuleMapString string) (map[int64]*SceneModule, error) {
- if len(sceneModuleMapString) <=0 {
+ if len(sceneModuleMapString) <= 0 {
return nil, fmt.Errorf("the param sceneModuleMapString:[%v] must not be empty", sceneModuleMapString)
}
@@ -69,24 +82,24 @@ func loadSceneModuleString(sceneModuleMapString string) (map[int64]*SceneModule,
}
func LoadWorkflowFromApollo(configParams map[string]string) (map[int64]*Workflow, error) {
- if configParams == nil || len(configParams)<=0 {
+ if configParams == nil || len(configParams) <= 0 {
return nil, fmt.Errorf("configParam is empty:%v", configParams)
}
//对每个文件逐个加到map
wfMap := make(map[int64]*Workflow)
for key, val := range configParams {
- if key == sceneKey {
+ if key == sceneKey || key == versionKey {
continue
}
wf, err := createWorkflowFromKV(key, val)
if wf == nil || err != nil {
- tlog.ErrorCount(context.TODO(),"loadWorkflowFromKV_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
+ tlog.Handler.ErrorCount(context.TODO(), "loadWorkflowFromKV_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
continue
}
wf.FlowCharts, err = NewWorkflowChart(wf.FlowChart)
if err != nil {
- tlog.ErrorCount(context.TODO(),"NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
+ tlog.Handler.ErrorCount(context.TODO(), "NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
continue
}
wfMap[wf.Id] = wf
@@ -97,7 +110,7 @@ func LoadWorkflowFromApollo(configParams map[string]string) (map[int64]*Workflow
func createWorkflowFromKV(key, val string) (*Workflow, error) {
strs := strings.Split(key, "-")
- if len(strs)<3 {
+ if len(strs) < 3 {
return nil, fmt.Errorf("invalid key:%v, it must start with :workflow-sceneId-workflowId", key)
}
@@ -105,25 +118,25 @@ func createWorkflowFromKV(key, val string) (*Workflow, error) {
if err != nil {
return nil, fmt.Errorf("invalid sceneId:%v in key:%v,err=%v", strs[1], key, err)
}
- workflowId, err := strconv.ParseInt(strs[2], 10, 64)
+ workflowId, err := strconv.ParseInt(strs[2], 10, 64)
if err != nil {
return nil, fmt.Errorf("invalid workflowId:%v in key:%v,err=%v", strs[2], key, err)
}
workflow := &Workflow{
- Id : workflowId,
- DimensionId : -1,
- SceneId : sceneId,
- FlowChart : val,
- FlowCharts : nil,
- FlowBranch : nil,
- IsDefault : 1,
- Range1 : dftRange,
- Range2 : "",
- Remark : "",
+ Id: workflowId,
+ DimensionId: -1,
+ SceneId: sceneId,
+ FlowChart: val,
+ FlowCharts: nil,
+ FlowBranch: nil,
+ IsDefault: 1,
+ Range1: dftRange,
+ Range2: "",
+ Remark: "",
//UpdateTime:,
- GroupName : "",
+ GroupName: "",
}
return workflow, nil
-}
\ No newline at end of file
+}
diff --git a/tg-core/wfengine/workflow_store.go b/wfengine/engine_from_file.go
similarity index 39%
rename from tg-core/wfengine/workflow_store.go
rename to wfengine/engine_from_file.go
index cb96a1ab21be01c1eedcbc38deb1b3111e9ab9d1..1b7fdc39c04959e5eb71ce5449941b7b174b1b9c 100644
--- a/tg-core/wfengine/workflow_store.go
+++ b/wfengine/engine_from_file.go
@@ -1,15 +1,16 @@
/**
- Description : loader of workflow config info from files
- Author : dayunzhangyunfeng@didiglobal.com
- Date : 2021-05-14
+Description : loader of workflow config info from files
+Author : dayunzhangyunfeng@didiglobal.com
+Date : 2021-05-14
*/
package wfengine
import (
"context"
+ "encoding/json"
"fmt"
- "github.com/didi/tg-flow/tg-core/common/tlog"
+ "github.com/didi/tg-flow/common/tlog"
"io/ioutil"
"os"
"strconv"
@@ -17,12 +18,69 @@ import (
)
const (
- defaultRange = "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99"
+ defaultRange = "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99"
+ versionFileName = "/version"
)
+func GetLatestVersionFromFile(path string) (string, error) {
+ jsonFile, err := os.Open(path + versionFileName)
+ if err != nil {
+ return "", fmt.Errorf("os.open error, file:%v, err:%v", path, err)
+ }
+ defer jsonFile.Close()
+
+ byteValue, err := ioutil.ReadAll(jsonFile)
+ if err != nil {
+ return "", fmt.Errorf("ioutil.ReadAll error, byteValue:%v,err=%v", string(byteValue), err)
+ }
+
+ return string(byteValue), nil
+}
+
+func NewWorkflowEngineFromFile(moduleObj ModuleObjBase, configPath string) (*WorkflowEngine, error) {
+ smMap, err := LoadSceneModuleMapFromFile(configPath + "/scene.json")
+ if err != nil {
+ return nil, err
+ }
+
+ //更新workflow
+ wfMap, err := LoadWorkflowFromFile(configPath, smMap)
+ if err != nil {
+ return nil, err
+ }
+
+ version, err1 := GetLatestVersionFromFile(configPath)
+ if err1 != nil {
+ tlog.Handler.ErrorCount(context.TODO(), "GetLatestVersionFromRedis_err", fmt.Sprintf("configPath=%v, err=%v", configPath, err1))
+ }
+
+ return NewWorkflowEngine(smMap, wfMap, version, moduleObj)
+}
+
+func LoadSceneModuleMapFromFile(path string) (map[int64]*SceneModule, error) {
+ jsonFile, err := os.Open(path)
+ if err != nil {
+ return nil, fmt.Errorf("os.open error, file:%v, err:%v", path, err)
+ }
+ defer jsonFile.Close()
+
+ byteValue, err := ioutil.ReadAll(jsonFile)
+ if err != nil {
+ return nil, fmt.Errorf("ioutil.ReadAll error, byteValue:%v,err=%v", string(byteValue), err)
+ }
+
+ var sceneModuleMap map[int64]*SceneModule
+ err = json.Unmarshal(byteValue, &sceneModuleMap)
+ if err != nil {
+ return nil, fmt.Errorf("json.Unmarshal error, byteValue:%v,err=%v", string(byteValue), err)
+ }
+
+ return sceneModuleMap, nil
+}
+
func LoadWorkflowFromFile(dirPath string, smMap map[int64]*SceneModule) (map[int64]*Workflow, error) {
//1.遍历目录
- filePaths:= make(map[string]string)
+ filePaths := make(map[string]string)
err := getWorkflowFiles(dirPath, filePaths)
if err != nil {
return nil, err
@@ -33,12 +91,12 @@ func LoadWorkflowFromFile(dirPath string, smMap map[int64]*SceneModule) (map[int
for filePath, _ := range filePaths {
wf, err := createWorkflowFromFile(filePath)
if wf == nil || err != nil {
- tlog.ErrorCount(context.TODO(),"loadWorkflowFromFile_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
+ tlog.Handler.ErrorCount(context.TODO(), "loadWorkflowFromFile_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
continue
}
wf.FlowCharts, err = NewWorkflowChart(wf.FlowChart)
if err != nil {
- tlog.ErrorCount(context.TODO(),"NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
+ tlog.Handler.ErrorCount(context.TODO(), "NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", wf, err))
continue
}
wfMap[wf.Id] = wf
@@ -48,8 +106,8 @@ func LoadWorkflowFromFile(dirPath string, smMap map[int64]*SceneModule) (map[int
}
func getWorkflowFiles(path string, filePaths map[string]string) error {
- if !strings.HasSuffix(path,"/"){
- path = path +"/"
+ if !strings.HasSuffix(path, "/") {
+ path = path + "/"
}
rd, err := ioutil.ReadDir(path)
@@ -60,9 +118,9 @@ func getWorkflowFiles(path string, filePaths map[string]string) error {
for _, fi := range rd {
if fi.IsDir() {
getWorkflowFiles(path+fi.Name(), filePaths)
- }else if strings.HasPrefix(fi.Name(),"workflow-") {
+ } else if strings.HasPrefix(fi.Name(), "workflow-") {
filePath := path + fi.Name()
- filePaths[filePath]= ""
+ filePaths[filePath] = ""
}
}
@@ -80,14 +138,14 @@ func createWorkflowFromFile(path string) (*Workflow, error) {
idx := strings.Index(fileName, ".")
fileName = fileName[:idx]
strs := strings.Split(fileName, "-")
- if len(strs)<3 {
- return nil, fmt.Errorf("invalid workflow file name:%v, it must start with :workflow-sceneId-workflowId",jsonFile.Name())
+ if len(strs) < 3 {
+ return nil, fmt.Errorf("invalid workflow file name:%v, it must start with :workflow-sceneId-workflowId", jsonFile.Name())
}
sceneId, err := strconv.ParseInt(strs[1], 10, 64)
if err != nil {
return nil, fmt.Errorf("invalid sceneId:%v in file name:%v,err=%v", strs[1], fileName, err)
}
- workflowId, err := strconv.ParseInt(strs[2], 10, 64)
+ workflowId, err := strconv.ParseInt(strs[2], 10, 64)
if err != nil {
return nil, fmt.Errorf("invalid workflowId:%v in file name:%v,err=%v", strs[2], fileName, err)
}
@@ -99,19 +157,19 @@ func createWorkflowFromFile(path string) (*Workflow, error) {
flowChartStr := string(byteValue)
workflow := &Workflow{
- Id : workflowId,
- DimensionId : -1,
- SceneId : sceneId,
- FlowChart : flowChartStr,
- FlowCharts : nil,
- FlowBranch : nil,
- IsDefault : 1,
- Range1 : defaultRange,
- Range2 : "",
- Remark : "",
+ Id: workflowId,
+ DimensionId: -1,
+ SceneId: sceneId,
+ FlowChart: flowChartStr,
+ FlowCharts: nil,
+ FlowBranch: nil,
+ IsDefault: 1,
+ Range1: defaultRange,
+ Range2: "",
+ Remark: "",
//UpdateTime:,
- GroupName : "",
+ GroupName: "",
}
return workflow, nil
-}
\ No newline at end of file
+}
diff --git a/tg-core/wfengine/workflow_dao.go b/wfengine/engine_from_redis.go
similarity index 38%
rename from tg-core/wfengine/workflow_dao.go
rename to wfengine/engine_from_redis.go
index a83a575b0033f8f7291ea2a5ad614131b92eb983..4762f895e341e5f743d96be1b953b5b936e441f2 100644
--- a/tg-core/wfengine/workflow_dao.go
+++ b/wfengine/engine_from_redis.go
@@ -10,19 +10,30 @@ import (
"context"
"encoding/json"
"fmt"
- "github.com/didi/tg-flow/tg-core/common/redis"
- "github.com/didi/tg-flow/tg-core/common/tlog"
+ "github.com/didi/tg-flow/common/tlog"
)
const (
- RedisKeyWorkflow = "workflow_app_"
+ RedisKeySceneModule = "scene_module_app_"
+ RedisKeyWorkflow = "workflow_app_"
+ RedisKeyVersion = "version_app_"
)
-func LoadWorkflow(appId int64, smMap map[int64]*SceneModule) (map[int64]*Workflow, error) {
- workflowMapStr, err := redis.Handler.Get(context.TODO(), fmt.Sprintf("%v%v",RedisKeyWorkflow, appId))
+//func GetLatestVersionFromRedis(appId int64) (string, error) {
+// return redis.Handler.Get(context.TODO(), fmt.Sprintf("%v%v", RedisKeyVersion, appId))
+//}
+
+func NewWorkflowEngineFromKV(moduleObj ModuleObjBase, sceneModuleMapString, workflowMapStr, version string) (*WorkflowEngine, error) {
+ var sceneModuleMap map[int64]*SceneModule
+ err := json.Unmarshal([]byte(sceneModuleMapString), &sceneModuleMap)
if err != nil {
- return nil, err
+ return nil, fmt.Errorf("err:%v, sceneModule:%v", err, sceneModuleMapString)
}
+ /*for sceneId, sceneModule := range sceneModuleMap {
+ if sceneModule.AppId != appId {
+ delete(sceneModuleMap, sceneId)
+ }
+ }*/
var workflowMap map[int64]*Workflow
err = json.Unmarshal([]byte(workflowMapStr), &workflowMap)
@@ -32,15 +43,15 @@ func LoadWorkflow(appId int64, smMap map[int64]*SceneModule) (map[int64]*Workflo
wfMap := make(map[int64]*Workflow)
for workflowId, workflow := range workflowMap {
- if _, ok := smMap[workflow.SceneId]; ok {
+ if _, ok := sceneModuleMap[workflow.SceneId]; ok {
workflow.FlowCharts, err = NewWorkflowChart(workflow.FlowChart)
if err != nil {
- tlog.ErrorCount(context.TODO(),"NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", workflow, err))
+ tlog.Handler.ErrorCount(context.TODO(), "NewWorkflowChart_err", fmt.Sprintf("wf:%v,err:%v", workflow, err))
continue
}
wfMap[workflowId] = workflow
}
}
- return wfMap, nil
-}
\ No newline at end of file
+ return NewWorkflowEngine(sceneModuleMap, wfMap, version, moduleObj)
+}
diff --git a/wfengine/flow_selector.go b/wfengine/flow_selector.go
new file mode 100644
index 0000000000000000000000000000000000000000..e29f9777f011566e1b5732d8d8a5ed320c0ae581
--- /dev/null
+++ b/wfengine/flow_selector.go
@@ -0,0 +1,15 @@
+package wfengine
+
+import "github.com/didi/tg-flow/model"
+
+/*
+*
+there are 3 flow selector:
+
+ 1: random
+ 2: custom
+ 3: apollo (apollo platform in didi)
+*/
+type FlowSelector interface {
+ SelectWorkflowId(sc *model.StrategyContext, sceneModule *SceneModule) (int64, string, error)
+}
diff --git a/wfengine/group_selector.go b/wfengine/group_selector.go
new file mode 100644
index 0000000000000000000000000000000000000000..8b4498bd5682cce3f54cf1e1f175058e3798e9e9
--- /dev/null
+++ b/wfengine/group_selector.go
@@ -0,0 +1,39 @@
+package wfengine
+
+import (
+ "fmt"
+ "github.com/didi/tg-flow/model"
+)
+
+type GroupSelector struct {
+ FlowSelector
+}
+
+/*
+*
+apollo分流,
+如果出现error，就取缺省分桶，同时返回error信息
+*/
+func (a *GroupSelector) SelectWorkflowId(sc *model.StrategyContext, sceneModule *SceneModule) (int64, string, error) {
+ //var err error
+ //if a.ApolloInfo == nil {
+ // return -1, "", fmt.Errorf("apollo info not initialized")
+ //}
+ //
+ //// 优先采用 ApolloInfo 中设置的分流实验名称，没有的话采用场景中配置的分流实验名称
+ //if a.ApolloInfo.GetDispatchExperimentName() == "" {
+ // a.ApolloInfo.SetDispatchExperimentName(sceneModule.DispatchExperimentName)
+ //}
+ //
+ //groupName, err := a.ApolloInfo.GetDispatchGroupName()
+ //if err != nil {
+ // return -1, groupName, fmt.Errorf("get dispatch groupName fail,groupName=%v, err=%v", groupName, err)
+ //}
+
+ workflowId, err := sceneModule.GetWorkflowId(sc.GroupName)
+ if err == nil {
+ return workflowId, sc.GroupName, nil
+ }
+
+ return -1, sc.GroupName, fmt.Errorf("select workflowId error, groupName=%v,workflowId=%v,err=%v", sc.GroupName, workflowId, err)
+}
diff --git a/tg-core/wfengine/inneraction/timeout_action.go b/wfengine/inneraction/timeout_action.go
similarity index 84%
rename from tg-core/wfengine/inneraction/timeout_action.go
rename to wfengine/inneraction/timeout_action.go
index ad34ae26e735c3ae2bce0c08c00c72732725ec84..87e4aa16921c2f8f82921b56b3c041ef2639ccf2 100644
--- a/tg-core/wfengine/inneraction/timeout_action.go
+++ b/wfengine/inneraction/timeout_action.go
@@ -2,8 +2,8 @@ package inneraction
import (
"context"
- "github.com/didi/tg-flow/tg-core/model"
- "github.com/didi/tg-flow/tg-core/wfengine"
+ "github.com/didi/tg-flow/model"
+ "github.com/didi/tg-flow/wfengine"
)
type TimeoutAction struct {
@@ -19,4 +19,4 @@ func (t TimeoutAction) DoAction(ctx context.Context, sc *model.StrategyContext)
func (m TimeoutAction) OnTimeout(context.Context, *model.StrategyContext) {
//this is a sample, you can do sth when timeout happen
//fmt.Println("default OnTimeout function, nothing to do !!!")
-}
\ No newline at end of file
+}
diff --git a/tg-core/wfengine/module_base.go b/wfengine/module_base.go
similarity index 72%
rename from tg-core/wfengine/module_base.go
rename to wfengine/module_base.go
index 2bad9b2a6cc46209e62318391465abf54a4d4350..409a1419db331531a2cfefe057da7f64a9514c72 100644
--- a/tg-core/wfengine/module_base.go
+++ b/wfengine/module_base.go
@@ -1,7 +1,7 @@
/**
- Description : ModelBase interface define
- Author : dayunzhangyunfeng@didiglobal.com
- Date : 2021-05-14
+Description : ModelBase interface define
+Author : dayunzhangyunfeng@didiglobal.com
+Date : 2021-05-14
*/
package wfengine
@@ -9,13 +9,13 @@ package wfengine
import (
"context"
"fmt"
- "github.com/didi/tg-flow/tg-core/common/tlog"
- "github.com/didi/tg-flow/tg-core/model"
+ "github.com/didi/tg-flow/common/tlog"
+ "github.com/didi/tg-flow/model"
"reflect"
"strconv"
)
-type IModelBase interface{
+type IModelBase interface {
DoAction(context.Context, *model.StrategyContext) interface{}
OnTimeout(context.Context, *model.StrategyContext)
SetName(string)
@@ -24,10 +24,10 @@ type IModelBase interface{
type ModelBase struct {
IModelBase
- Name string
+ Name string
}
-func (m *ModelBase) DoAction(context.Context, *model.StrategyContext) interface{}{
+func (m *ModelBase) DoAction(context.Context, *model.StrategyContext) interface{} {
return nil
}
@@ -44,24 +44,29 @@ func (m *ModelBase) GetName() string {
}
type ModuleObjBase interface {
- NewObj(moduleName string, vMap map[string]string) IModelBase
+ NewObj(moduleName string) IModelBase
}
-func ReflectModuleField(obj interface{}, reflectType reflect.Type, vMap map[string]string) error {
+func reflectModuleField(obj interface{}, vMap map[string]string) error {
+ if len(vMap) == 0 {
+ return nil
+ }
+
v := reflect.ValueOf(obj)
if v.Kind() == reflect.Ptr && !v.Elem().CanSet() {
err := fmt.Errorf("this obj is not match reflect")
- tlog.ErrorCount(context.TODO(),"ReflectModuleField_err", fmt.Sprintf("obj:%v, err:%v", obj, err))
+ tlog.Handler.ErrorCount(context.TODO(), "ReflectModuleField_err", fmt.Sprintf("obj:%v, err:%v", obj, err))
return err
}
+ reflectType := reflect.Indirect(v).Type()
v = v.Elem()
for i := 0; i < reflectType.NumField(); i++ {
field := reflectType.Field(i)
fieldValue := v.FieldByName(field.Name)
if !fieldValue.IsValid() {
err := fmt.Errorf("this obj(" + fmt.Sprintf("%v", obj) + ") field(" + field.Name + ")")
- tlog.ErrorCount(context.TODO(),"ReflectModuleField_err", fmt.Sprintf("%v", err))
+ tlog.Handler.ErrorCount(context.TODO(), "ReflectModuleField_err", fmt.Sprintf("%v", err))
continue
}
@@ -76,7 +81,7 @@ func ReflectModuleField(obj interface{}, reflectType reflect.Type, vMap map[stri
tempInt, err := strconv.ParseInt(vMap[field.Name], 10, 64)
if err != nil {
err := fmt.Errorf("obj(" + fmt.Sprintf("%v", obj) + ") field(" + field.Name + ")'s value(" + vMap[field.Name] + ")")
- tlog.ErrorCount(context.TODO(),"ReflectModuleField_err", fmt.Sprintf("%v", err))
+ tlog.Handler.ErrorCount(context.TODO(), "ReflectModuleField_err", fmt.Sprintf("%v", err))
continue
}
fieldValue.SetInt(tempInt)
@@ -85,7 +90,7 @@ func ReflectModuleField(obj interface{}, reflectType reflect.Type, vMap map[stri
tempFolat, err := strconv.ParseFloat(vMap[field.Name], 64)
if err != nil {
err := fmt.Errorf("obj(" + fmt.Sprintf("%v", obj) + ") field(" + field.Name + ")'s value(" + vMap[field.Name] + ")")
- tlog.ErrorCount(context.TODO(),"ReflectModuleField_err", fmt.Sprintf("%v", err))
+ tlog.Handler.ErrorCount(context.TODO(), "ReflectModuleField_err", fmt.Sprintf("%v", err))
continue
}
fieldValue.SetFloat(tempFolat)
diff --git a/tg-core/wfengine/flow_splitter.go b/wfengine/random_selector.go
similarity index 30%
rename from tg-core/wfengine/flow_splitter.go
rename to wfengine/random_selector.go
index 01e53390da3af992e36480306e6d5c951d4e34b1..c7850c05a2947916e5312d4f3e8ce01c62f97e23 100644
--- a/tg-core/wfengine/flow_splitter.go
+++ b/wfengine/random_selector.go
@@ -1,24 +1,28 @@
-/**
-Description : flow splitter
-Author : dayunzhangyunfeng@didiglobal.com
-Date : 2021-05-14
-*/
package wfengine
import (
"fmt"
- "github.com/didi/tg-flow/tg-core/model"
+ "github.com/didi/tg-flow/model"
"hash/crc32"
+ "strconv"
"time"
)
-//在线随机分流
-func FlowByOnlineRandom(sc *model.StrategyContext, sceneModule *SceneModule) (int64, int) {
+type RandomSelector struct {
+ FlowSelector
+}
+
+// 在线随机分流
+func (r *RandomSelector) SelectWorkflowId(sc *model.StrategyContext, sceneModule *SceneModule) (int64, string, error) {
//1、根据用户id，算出一个0-99的数字
slotId := getSlotId(sc.UserId, sceneModule.BucketType)
-
//2、在对应的维度id内，根据slotId，选择对应的workflow
- return sceneModule.SlotMap[slotId], slotId
+ workflowId, ok := sceneModule.SlotMap[slotId]
+ if !ok {
+ return -1, "", fmt.Errorf("no workflowId found, UserId=%v, BucketType=%v", sc.UserId, sceneModule.BucketType)
+ }
+
+ return workflowId, strconv.Itoa(slotId), nil
}
func getSlotId(str string, bucketType int) int {
@@ -32,28 +36,3 @@ func getSlotId(str string, bucketType int) int {
}
return v % 100
}
-
-// FlowByApollo apollo分流
-func FlowByApollo(sc *model.StrategyContext, sceneModule *SceneModule) (int64, string, error) {
- var err error
- if sc.ApolloInfo == nil {
- return 0, "", fmt.Errorf("apollo info not initialized")
- }
-
- // 优先采用 ApolloInfo 中设置的分流实验名称，没有的话采用场景中配置的分流实验名称
- if sc.ApolloInfo.GetDispatchExperimentName() == "" {
- sc.ApolloInfo.SetDispatchExperimentName(sceneModule.DispatchExperimentName)
- }
-
- groupName, err := sc.ApolloInfo.GetDispatchGroupName()
- if err != nil {
- return 0, "", err
- }
-
- workflowId, err := sceneModule.GetWorkflowId(groupName)
- if err == nil {
- return workflowId, groupName, nil
- } else {
- return workflowId, groupName, err
- }
-}
diff --git a/tg-core/wfengine/scene_module.go b/wfengine/scene_module.go
similarity index 100%
rename from tg-core/wfengine/scene_module.go
rename to wfengine/scene_module.go
diff --git a/tg-core/wfengine/test/condition_external.go b/wfengine/test/condition_external.go
similarity index 100%
rename from tg-core/wfengine/test/condition_external.go
rename to wfengine/test/condition_external.go
diff --git a/tg-core/wfengine/time_waiter.go b/wfengine/time_waiter.go
similarity index 100%
rename from tg-core/wfengine/time_waiter.go
rename to wfengine/time_waiter.go
diff --git a/tg-core/wfengine/workflow.go b/wfengine/workflow.go
similarity index 86%
rename from tg-core/wfengine/workflow.go
rename to wfengine/workflow.go
index f580d08be523dcd83c734d9fc20f5c951cc70bc4..150ee53ed23fb4bf42e858cb3b8c7e8d56e18990 100644
--- a/tg-core/wfengine/workflow.go
+++ b/wfengine/workflow.go
@@ -87,14 +87,14 @@ func NewWorkflowChart(flowChartStr string) (*WorkflowChart, error) {
if len(flowChartStr) == 0 {
return nil, errors.New("create WorkflowChart fail, flowChartStr is empty")
}
-
+
flowChart := &WorkflowChart{}
//读取的数据为json格式，需要进行解码
err := json.Unmarshal([]byte(flowChartStr), flowChart)
if err != nil {
return nil, fmt.Errorf("create WorkflowChart fail, invalid json:%v, err:%v", flowChartStr, err)
}
-
+
err = flowChart.setFirstActionId()
if err != nil {
return nil, fmt.Errorf("create WorkflowChart fail, err:%v", err)
@@ -114,7 +114,7 @@ func (w *WorkflowChart) setPrevActionIds(actionId string){
if actionId == w.FirstActionId {
action.PrevActionIds = []string{}
}
-
+
nextActionIds := action.NextActionIds
for _, nextActionId := range nextActionIds {
nextAction, ok := w.ActionMap[nextActionId]
@@ -122,7 +122,7 @@ func (w *WorkflowChart) setPrevActionIds(actionId string){
//报个error？
continue
}
-
+
if nextAction.PrevActionIds == nil || len(nextAction.PrevActionIds) == 0 {
nextAction.PrevActionIds = []string{actionId}
}else{
@@ -138,7 +138,7 @@ func (w *WorkflowChart) setPrevActionIds(actionId string){
nextAction.PrevActionIds = append(nextAction.PrevActionIds, actionId)
}
}
-
+
w.setPrevActionIds(nextActionId)
}
}
@@ -153,19 +153,19 @@ func (w *WorkflowChart) setFirstActionId() error {
w.LastActionId = action.ActionId
continue
}
-
+
for _, nextId := range action.NextActionIds {
nextActionIds[nextId] = true
}
}
-
+
for actionId,_ := range w.ActionMap {
if _, ok := nextActionIds[actionId]; !ok {
w.FirstActionId = actionId
return nil
}
}
-
+
return errors.New("first action not found")
}
@@ -184,7 +184,7 @@ func (w *WorkflowChart) CreateWaitMap() (map[string]*sync.WaitGroup, map[string]
tsMap[actionId] = NewTimeWaiter(action.Timeout)
}
}
-
+
return wgMap,tsMap
}
@@ -250,30 +250,44 @@ func (a *Action) createParamSlice(paramValues *sync.Map) ([]interface{}, error)
}
func (a *Action) Detach(prevAction *Action) {
- if prevAction == nil {
+ if prevAction == nil || len(a.PrevActionIds)==0 {
//todo error
return
}
- pActionIds := make([]string, len(a.PrevActionIds)-1)
- previ := 0
- for _, prevActionId := range a.PrevActionIds {
- if prevActionId != prevAction.ActionId {
- pActionIds[previ] = prevActionId
- previ++
+ prevId := -1
+ for pId, prevActionId := range a.PrevActionIds {
+ if prevActionId == prevAction.ActionId {
+ prevId = pId
+ break
}
}
- a.PrevActionIds = pActionIds
+ if prevId > -1{
+ a.PrevActionIds = append(a.PrevActionIds[:prevId], a.PrevActionIds[prevId+1:]...)
+ }
- nActionIds := make([]string, len(prevAction.NextActionIds)-1)
- nexti := 0
- for _, nextActionId := range prevAction.NextActionIds {
- if nextActionId != a.ActionId {
- nActionIds[nexti] = nextActionId
- nexti++
+ nextId := -1
+ for nId, nextActionId := range prevAction.NextActionIds {
+ if nextActionId == a.ActionId {
+ nextId = nId
+ break
+ }
+ }
+ //fmt.Println("nextId============>", nextId)
+ //fmt.Println("before set prevAction.NextActionIds===>", strings.Join(prevAction.NextActionIds,","))
+ if nextId>-1 {
+ prevAction.NextActionIds = append(prevAction.NextActionIds[:nextId], prevAction.NextActionIds[nextId+1:]...)
+ //fmt.Println("after set prevAction.NextActionIds===>", strings.Join(prevAction.NextActionIds,","))
+ if len(prevAction.NextConditions)> nextId {
+ prevAction.NextConditions = append(prevAction.NextConditions[:nextId], prevAction.NextConditions[nextId+1:]...)
}
+ //fmt.Println("after set prevAction.NextConditions===>", strings.Join(prevAction.NextConditions,","))
}
- prevAction.NextActionIds = nActionIds
+ //fmt.Println("\nfinish detach, prevActionId, actionId:", prevAction.ActionId, a.ActionId)
+}
+
+func (a *Action) toString() string {
+ return fmt.Sprintf("actionId:%+v, nextActionIds:%+v, nextConditions:%+v, prevActionIds:%+v",a.ActionId, strings.Join(a.NextActionIds,","),strings.Join(a.NextConditions,","))
}
func (a *Action) clone() *Action {
@@ -306,6 +320,7 @@ func (w *WorkflowChart) clone() *WorkflowChart {
chart := &WorkflowChart{}
chart.FirstActionId = w.FirstActionId
chart.LastActionId = w.LastActionId
+ chart.HashCondition = w.HashCondition
actionMap := make(map[string]*Action)
for actionId, action := range w.ActionMap {
@@ -329,6 +344,7 @@ func copyStringArray(sources []string) []string {
}
func (a *Action) executeCond(paramValues *sync.Map) (retActionId string, err error) {
+ //fmt.Println("\n\n\n开始执行executeCond:",fmt.Sprintf("action:%+v", a))
//若报错，有缺省用缺省,无缺省用最后一个。条件必含后继，配置保存时校验
defaultIndex := len(a.NextActionIds) - 1
for idx, cdt := range a.NextConditions {
@@ -336,8 +352,10 @@ func (a *Action) executeCond(paramValues *sync.Map) (retActionId string, err err
defaultIndex = idx
}
}
+ //fmt.Println("defaultIndex=======>", defaultIndex)
retActionId = a.NextActionIds[defaultIndex]
+ //fmt.Println("retActionId=>", retActionId)
err = nil
defer func() {
if err0 := recover(); err0 != nil {
@@ -350,8 +368,9 @@ func (a *Action) executeCond(paramValues *sync.Map) (retActionId string, err err
err = fmt.Errorf("createParamSlice error, default value: %v used, a.Params:%v, paramValues:%v, err:%v", a.NextActionIds[defaultIndex], a.Params, paramValues, err)
return
}
-
+ //fmt.Println("prepare to exe:", a.ActionName, params)
val, err := GetCondExecutors().Execute(a.ActionName, params)
+ //fmt.Println("条件执行结果 result: val====>", val, "err====>", err)
if err != nil {
err = fmt.Errorf("GetCondExecutors().Execute error, default value: %v used, actionName:%v, params:%v, err:%v", a.NextActionIds[defaultIndex], a.ActionName, params, err)
return
@@ -389,25 +408,6 @@ func (w *WorkflowChart) newWorkflowBranch() *WorkflowBranch {
}
}
-/*func (w *WorkflowBranch) selectBranchKey(params map[string]interface{}) (string, error) {
- if !w.hasBranch() {
- return BranchKeyDefault, nil
- }
-
- branchMap := make(map[string]int)
- for _, actionId := range w.SortedBranch {
- idx, err := w.ActionMap[actionId].executeCond(params)
- if err != nil {
- return "", fmt.Errorf("execute cond error:%v", err)
- }
- branchMap[actionId] = idx
- }
-
- branchKey := w.getBranchKey(branchMap)
-
- return branchKey, nil
-}*/
-
func (w *WorkflowBranch) getCurrentBranchKey() string {
return w.getBranchKey(w.CurrentBranch)
}
diff --git a/tg-core/wfengine/workflow.json b/wfengine/workflow.json
similarity index 100%
rename from tg-core/wfengine/workflow.json
rename to wfengine/workflow.json
-