add Action module

flow/kis_flow_action.go

@@ -0,0 +1,63 @@
+package flow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"kis-flow/kis"
+)
+
+// dealAction  处理Action，决定接下来Flow的流程走向
+func (flow *KisFlow) dealAction(ctx context.Context, fn kis.Function) (kis.Function, error) {
+
+	// DataReuse Action
+	if flow.action.DataReuse {
+		if err := flow.commitReuseData(ctx); err != nil {
+			return nil, err
+		}
+	} else {
+		if err := flow.commitCurData(ctx); err != nil {
+			return nil, err
+		}
+	}
+
+	// ForceEntryNext Action
+	if flow.action.ForceEntryNext {
+		if err := flow.commitVoidData(ctx); err != nil {
+			return nil, err
+		}
+		flow.abort = false
+	}
+
+	// JumpFunc Action
+	if flow.action.JumpFunc != "" {
+		if _, ok := flow.Funcs[flow.action.JumpFunc]; !ok {
+			//当前JumpFunc不在flow中
+			return nil, errors.New(fmt.Sprintf("Flow Jump -> %s is not in Flow", flow.action.JumpFunc))
+		}
+
+		jumpFunction := flow.Funcs[flow.action.JumpFunc]
+		// 更新上层Function
+		flow.PrevFunctionId = jumpFunction.GetPrevId()
+		fn = jumpFunction
+
+		// 如果设置跳跃，强制跳跃
+		flow.abort = false
+
+	} else {
+
+		// 更新上一层 FuncitonId 游标
+		flow.PrevFunctionId = flow.ThisFunctionId
+		fn = fn.Next()
+	}
+
+	// Abort Action 强制终止
+	if flow.action.Abort {
+		flow.abort = true
+	}
+
+	// 清空Action
+	flow.action = kis.Action{}
+
+	return fn, nil
+}

kis/action.go

@@ -0,0 +1,58 @@
+package kis
+
+// Action KisFlow执行流程Actions
+type Action struct {
+	// DataReuse 是否复用上层Function数据
+	DataReuse bool
+
+	// 默认Next()为如果本层Function计算结果为0条数据，之后Function将不会继续执行
+	// ForceEntryNext 为忽略上述默认规则，没有数据强制进入下一层Function
+	ForceEntryNext bool
+
+	// JumpFunc 跳转到指定Function继续执行
+	JumpFunc string
+
+	// Abort 终止Flow的执行
+	Abort bool
+}
+
+// ActionFunc KisFlow Functional Option 类型
+type ActionFunc func(ops *Action)
+
+// LoadActions 加载Actions，依次执行ActionFunc操作函数
+func LoadActions(acts []ActionFunc) Action {
+	action := Action{}
+
+	if acts == nil {
+		return action
+	}
+
+	for _, act := range acts {
+		act(&action)
+	}
+
+	return action
+}
+
+// ActionDataReuse Next复用上层Function数据Option
+func ActionDataReuse(act *Action) {
+	act.DataReuse = true
+}
+
+// ActionForceEntryNext 强制进入下一层
+func ActionForceEntryNext(act *Action) {
+	act.ForceEntryNext = true
+}
+
+// ActionJumpFunc 会返回一个ActionFunc函数，并且会将funcName赋值给Action.JumpFunc
+// (注意：容易出现Flow循环调用，导致死循环)
+func ActionJumpFunc(funcName string) ActionFunc {
+	return func(act *Action) {
+		act.JumpFunc = funcName
+	}
+}
+
+// ActionAbort 终止Flow的执行
+func ActionAbort(action *Action) {
+	action.Abort = true
+}

test/faas/faas_abort.go

@@ -0,0 +1,20 @@
+package faas
+
+import (
+	"context"
+	"fmt"
+	"kis-flow/kis"
+)
+
+// type FaaS func(context.Context, Flow) error
+
+func AbortFuncHandler(ctx context.Context, flow kis.Flow) error {
+	fmt.Println("---> Call AbortFuncHandler ----")
+
+	for _, row := range flow.Input() {
+		str := fmt.Sprintf("In FuncName = %s, FuncId = %s, row = %s", flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
+		fmt.Println(str)
+	}
+
+	return flow.Next(kis.ActionAbort)
+}

test/faas/faas_data_reuse.go

@@ -0,0 +1,26 @@
+package faas
+
+import (
+	"context"
+	"fmt"
+	"kis-flow/kis"
+)
+
+// type FaaS func(context.Context, Flow) error
+
+func DataReuseFuncHandler(ctx context.Context, flow kis.Flow) error {
+	fmt.Println("---> Call DataReuseFuncHandler ----")
+
+	for index, row := range flow.Input() {
+		str := fmt.Sprintf("In FuncName = %s, FuncId = %s, row = %s", flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
+		fmt.Println(str)
+
+		// 计算结果数据
+		resultStr := fmt.Sprintf("data from funcName[%s], index = %d", flow.GetThisFuncConf().FName, index)
+
+		// 提交结果数据
+		_ = flow.CommitRow(resultStr)
+	}
+
+	return flow.Next(kis.ActionDataReuse)
+}

test/faas/faas_jump.go

@@ -0,0 +1,20 @@
+package faas
+
+import (
+	"context"
+	"fmt"
+	"kis-flow/kis"
+)
+
+// type FaaS func(context.Context, Flow) error
+
+func JumpFuncHandler(ctx context.Context, flow kis.Flow) error {
+	fmt.Println("---> Call JumpFuncHandler ----")
+
+	for _, row := range flow.Input() {
+		str := fmt.Sprintf("In FuncName = %s, FuncId = %s, row = %s", flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
+		fmt.Println(str)
+	}
+
+	return flow.Next(kis.ActionJumpFunc("funcName1"))
+}

test/faas/faas_no_result.go

@@ -0,0 +1,20 @@
+package faas
+
+import (
+	"context"
+	"fmt"
+	"kis-flow/kis"
+)
+
+// type FaaS func(context.Context, Flow) error
+
+func NoResultFuncHandler(ctx context.Context, flow kis.Flow) error {
+	fmt.Println("---> Call NoResultFuncHandler ----")
+
+	for _, row := range flow.Input() {
+		str := fmt.Sprintf("In FuncName = %s, FuncId = %s, row = %s", flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
+		fmt.Println(str)
+	}
+
+	return flow.Next(kis.ActionForceEntryNext)
+}

test/kis_action_test.go

@@ -0,0 +1,135 @@
+package test
+
+import (
+	"context"
+	"kis-flow/common"
+	"kis-flow/file"
+	"kis-flow/kis"
+	"kis-flow/test/caas"
+	"kis-flow/test/faas"
+	"testing"
+)
+
+func TestActionAbort(t *testing.T) {
+	ctx := context.Background()
+
+	// 0. 注册Function 回调业务
+	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
+	kis.Pool().FaaS("abortFunc", faas.AbortFuncHandler) // 添加abortFunc 业务
+	kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)
+
+	// 0. 注册ConnectorInit 和 Connector 回调业务
+	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
+	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)
+
+	// 1. 加载配置文件并构建Flow
+	if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
+		panic(err)
+	}
+
+	// 2. 获取Flow
+	flow1 := kis.Pool().GetFlow("flowName2")
+
+	// 3. 提交原始数据
+	_ = flow1.CommitRow("This is Data1 from Test")
+	_ = flow1.CommitRow("This is Data2 from Test")
+	_ = flow1.CommitRow("This is Data3 from Test")
+
+	// 4. 执行flow1
+	if err := flow1.Run(ctx); err != nil {
+		panic(err)
+	}
+}
+
+func TestActionDataReuse(t *testing.T) {
+	ctx := context.Background()
+
+	// 0. 注册Function 回调业务
+	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
+	kis.Pool().FaaS("dataReuseFunc", faas.DataReuseFuncHandler) // 添加dataReuesFunc 业务
+	kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)
+
+	// 0. 注册ConnectorInit 和 Connector 回调业务
+	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
+	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)
+
+	// 1. 加载配置文件并构建Flow
+	if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
+		panic(err)
+	}
+
+	// 2. 获取Flow
+	flow1 := kis.Pool().GetFlow("flowName3")
+
+	// 3. 提交原始数据
+	_ = flow1.CommitRow("This is Data1 from Test")
+	_ = flow1.CommitRow("This is Data2 from Test")
+	_ = flow1.CommitRow("This is Data3 from Test")
+
+	// 4. 执行flow1
+	if err := flow1.Run(ctx); err != nil {
+		panic(err)
+	}
+}
+
+func TestActionForceEntry(t *testing.T) {
+	ctx := context.Background()
+
+	// 0. 注册Function 回调业务
+	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
+	kis.Pool().FaaS("noResultFunc", faas.NoResultFuncHandler) // 添加noResultFunc 业务
+	kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)
+
+	// 0. 注册ConnectorInit 和 Connector 回调业务
+	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
+	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)
+
+	// 1. 加载配置文件并构建Flow
+	if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
+		panic(err)
+	}
+
+	// 2. 获取Flow
+	flow1 := kis.Pool().GetFlow("flowName4")
+
+	// 3. 提交原始数据
+	_ = flow1.CommitRow("This is Data1 from Test")
+	_ = flow1.CommitRow("This is Data2 from Test")
+	_ = flow1.CommitRow("This is Data3 from Test")
+
+	// 4. 执行flow1
+	if err := flow1.Run(ctx); err != nil {
+		panic(err)
+	}
+}
+
+func TestActionJumpFunc(t *testing.T) {
+	ctx := context.Background()
+
+	// 0. 注册Function 回调业务
+	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
+	kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler)
+	kis.Pool().FaaS("jumpFunc", faas.JumpFuncHandler) // 添加jumpFunc 业务
+
+	// 0. 注册ConnectorInit 和 Connector 回调业务
+	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
+	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)
+
+	// 1. 加载配置文件并构建Flow
+	if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
+		panic(err)
+	}
+
+	// 2. 获取Flow
+	flow1 := kis.Pool().GetFlow("flowName5")
+
+	// 3. 提交原始数据
+	_ = flow1.CommitRow("This is Data1 from Test")
+	_ = flow1.CommitRow("This is Data2 from Test")
+	_ = flow1.CommitRow("This is Data3 from Test")
+
+	// 4. 执行flow1
+	if err := flow1.Run(ctx); err != nil {
+		panic(err)
+	}
+}

test/load_conf/flow/flow-FlowName2.yml

@@ -0,0 +1,7 @@
+kistype: flow
+status: 1
+flow_name: flowName2
+flows:
+  - fname: funcName1
+  - fname: abortFunc
+  - fname: funcName3

test/load_conf/flow/flow-FlowName3.yml

@@ -0,0 +1,7 @@
+kistype: flow
+status: 1
+flow_name: flowName3
+flows:
+  - fname: funcName1
+  - fname: dataReuseFunc
+  - fname: funcName3

test/load_conf/flow/flow-FlowName4.yml

@@ -0,0 +1,7 @@
+kistype: flow
+status: 1
+flow_name: flowName4
+flows:
+  - fname: funcName1
+  - fname: noResultFunc
+  - fname: funcName3

test/load_conf/flow/flow-FlowName5.yml

@@ -0,0 +1,7 @@
+kistype: flow
+status: 1
+flow_name: flowName5
+flows:
+  - fname: funcName1
+  - fname: funcName2
+  - fname: jumpFunc

test/load_conf/func/func-AbortFunc.yml

@@ -0,0 +1,8 @@
+kistype: func
+fname: abortFunc
+fmode: Calculate
+source:
+  name: 用户订单错误率
+  must:
+    - order_id
+    - user_id

test/load_conf/func/func-NoResultFunc.yml

@@ -0,0 +1,8 @@
+kistype: func
+fname: noResultFunc
+fmode: Calculate
+source:
+  name: 用户订单错误率
+  must:
+    - order_id
+    - user_id

test/load_conf/func/func-dataReuseFunc.yml

@@ -0,0 +1,8 @@
+kistype: func
+fname: dataReuseFunc
+fmode: Calculate
+source:
+  name: 用户订单错误率
+  must:
+    - order_id
+    - user_id

test/load_conf/func/func-jumpFunc.yml

@@ -0,0 +1,8 @@
+kistype: func
+fname: jumpFunc
+fmode: Calculate
+source:
+  name: 用户订单错误率
+  must:
+    - order_id
+    - user_id