Golang框架实战-KisFlow流式计算框架(8)-KisFlow Action-六虎

7.1 Action Abort（停止流程）

KisFlow Action 是指在履行Function的时候，一起能够操控Flow的调度逻辑，KisFlow提供一些Action动作让开发者做选择，本节先介绍最简单的Action动作，Abort（停止当时Flow）。

咱们终究的Abort的运用形式如下：

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)  // 停止Flow
}

AbortFuncHandler()是一个Function 的事务回调办法，是由开发者自界说的，在履行完当时Funciton之后，正常的情况是持续履行下一个Funciton，可是假如传递flow.Next(kis.ActionAbort) 作为当时Funciton的回来值，那么则不会履行到下一个Funciton,而是直接停止当时Flow的调度核算流。

下面咱们先来完成KisFlow的 Abort Action动作形式。

7.1.1 Abort接口界说

首要，先对Flow的Abort()接口做界说。

kis-flow/kis/flow.go

type Flow interface {
	// Run 调度Flow，顺次调度Flow中的Function而且履行
	Run(ctx context.Context) error
	// Link 将Flow中的Function按照装备文件中的装备进行连接
	Link(fConf *config.KisFuncConfig, fParams config.FParam) error
	// CommitRow 提交Flow数据到即将履行的Function层
	CommitRow(row interface{}) error
	// Input 得到flow当时履行Function的输入源数据
	Input() common.KisRowArr
	// GetName 得到Flow的称号
	GetName() string
	// GetThisFunction 得到当时正在履行的Function
	GetThisFunction() Function
	// GetThisFuncConf 得到当时正在履行的Function的装备
	GetThisFuncConf() *config.KisFuncConfig
	// GetConnector 得到当时正在履行的Function的Connector
	GetConnector() (Connector, error)
	// GetConnConf 得到当时正在履行的Function的Connector的装备
	GetConnConf() (*config.KisConnConfig, error)
	// GetConfig 得到当时Flow的装备
	GetConfig() *config.KisFlowConfig
	// GetFuncConfigByName 得到当时Flow的装备
	GetFuncConfigByName(funcName string) *config.KisFuncConfig
	//  --- KisFlow Action ---
	// Next 当时Flow履行到的Function进入下一层Function所带着的Action动作
	Next(acts ...ActionFunc) error
}

这儿边提供一个接口Next(acts ...ActionFunc) error，其间参数是一个可变参数，类型为ActionFunc，这个是咱们给KisFlow界说的Action相关的办法。有关Action的界说模块如下：

7.1.2 Action模块界说

Action是用来在Flow履行过程中，经过Function来操控Flow履行特别动作的行为装备模块，包括上面的Abort行为，Abort也归于其间一个Action。Action的模块界说如下，在kis-flow/kis/下创立action.go文件，完成：

kis-flow/kis/action.go

package kis
// Action KisFlow履行流程Actions
type Action struct {
	// 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
}
// ActionAbort 停止Flow的履行
func ActionAbort(action *Action) {
	action.Abort = true
}

首要，现在Action只要Abort一个行为，咱们用bool类型来表明Abort是否为停止，true则为需求停止flow的调用。
其次，type ActionFunc func(ops *Action)这个函数原型为一个函数类型，函数的形参是传递进来一个Action{} 指针，而 func ActionAbort(action *Action)则是它的一个详细的函数，ActionAbort()的办法的意图便是将Action的Abort成员设置为true。

最终看func LoadActions(acts []ActionFunc) Action办法。这个形参是一个ActionFunc函数数组，LoadActions()则是创立一个新的Action{} ，然后顺次履行[]ActionFunc的函数来改动Aciton{}的成员，终究将新的Action{}回来上层。

7.1.3 Next办法完成

接下来，咱们需求给KisFlow模块完成这个接口，首要需求给KisFlow增加一个Action{}成员，表明每次履行完Function之后所带着的动作。

kis-flow/flow/kis_flow.go

// KisFlow 用于贯穿整条流式核算的上下文环境
type KisFlow struct {
	// 根底信息
	Id   string                // Flow的分布式实例ID(用于KisFlow内部区别不同实例)
	Name string                // Flow的可读称号
	Conf *config.KisFlowConfig // Flow装备策略
	// Function列表
	Funcs          map[string]kis.Function // 当时flow具有的全部办理的全部Function目标, key: FunctionName
	FlowHead       kis.Function            // 当时Flow所具有的Function列表表头
	FlowTail       kis.Function            // 当时Flow所具有的Function列表表尾
	flock          sync.RWMutex            // 办理链表刺进读写的锁
	ThisFunction   kis.Function            // Flow当时正在履行的KisFunction目标
	ThisFunctionId string                  // 当时履行到的Function ID
	PrevFunctionId string                  // 当时履行到的Function 上一层FunctionID
	// Function列表参数
	funcParams map[string]config.FParam // flow在当时Function的自界说固定装备参数,Key:function的实例KisID, value:FParam
	fplock     sync.RWMutex             // 办理funcParams的读写锁
	// 数据
	buffer common.KisRowArr  // 用来临时寄存输入字节数据的内部Buf, 一条数据为interface{}, 多条数据为[]interface{} 也便是KisBatch
	data   common.KisDataMap // 流式核算各个层级的数据源
	inPut  common.KisRowArr  // 当时Function的核算输入数据
    // +++++++++++++++++++++
	// KisFlow Action
	action kis.Action        // 当时Flow所带着的Action动作
}

然后完成KisFlow的Next()接口，如下：

kis-flow/flow/kis_flow.go

// Next 当时Flow履行到的Function进入下一层Function所带着的Action动作
func (flow *KisFlow) Next(acts ...kis.ActionFunc) error {
	// 加载Function FaaS 传递的 Action动作
	flow.action = kis.LoadActions(acts)
	return nil
}

每次开发者在履行Function的自界说事务回调中，最终会调用flow.Next()来传递Action，所以 Next(acts ...kis.ActionFunc) error便是讲传递的Action特点加载进来而且在flow.action保存。

7.1.4 Abort操控Flow流程

现在有个Abort来操控Flow流，那么咱们需求给KisFlow增加一个成员来表明这个状况

kis-flow/flow/kis_flow.go

// KisFlow 用于贯穿整条流式核算的上下文环境
type KisFlow struct {
	// 根底信息
	Id   string                // Flow的分布式实例ID(用于KisFlow内部区别不同实例)
	Name string                // Flow的可读称号
	Conf *config.KisFlowConfig // Flow装备策略
	// Function列表
	Funcs          map[string]kis.Function // 当时flow具有的全部办理的全部Function目标, key: FunctionName
	FlowHead       kis.Function            // 当时Flow所具有的Function列表表头
	FlowTail       kis.Function            // 当时Flow所具有的Function列表表尾
	flock          sync.RWMutex            // 办理链表刺进读写的锁
	ThisFunction   kis.Function            // Flow当时正在履行的KisFunction目标
	ThisFunctionId string                  // 当时履行到的Function ID
	PrevFunctionId string                  // 当时履行到的Function 上一层FunctionID
	// Function列表参数
	funcParams map[string]config.FParam // flow在当时Function的自界说固定装备参数,Key:function的实例KisID, value:FParam
	fplock     sync.RWMutex             // 办理funcParams的读写锁
	// 数据
	buffer common.KisRowArr  // 用来临时寄存输入字节数据的内部Buf, 一条数据为interface{}, 多条数据为[]interface{} 也便是KisBatch
	data   common.KisDataMap // 流式核算各个层级的数据源
	inPut  common.KisRowArr  // 当时Function的核算输入数据
	action kis.Action        // 当时Flow所带着的Action动作
    // +++++++++
    abort  bool              // 是否中断Flow
}

在每次履行到flow.Run()办法时，需求重置abort变量，而且在循环调度的时候加上对flow.abort的判别。

kis-flow/flow/kis_flow.go

// Run 发动KisFlow的流式核算, 从开端Function开端履行流
func (flow *KisFlow) Run(ctx context.Context) error {
    // +++++++++
	// 重置 abort
	flow.abort = false  //  每次进入调度，要重置abort状况
	// ... ...
    // ... ...
	//流式链式调用
	for fn != nil && flow.abort != true { // ++++ 假如设置abort则不进入下次循环调度
		// ... ...
        // ... ...
        if err := fn.Call(ctx, flow); err != nil {
			//Error
			return err
		} else {
			//Success
			// ... ...
			fn = fn.Next()
        }
	}
	return nil

这样在每次Call()调度到Funciton的自定办法时，假如return flow.Next(ActionAbort)就会对flow的Action状况进行改动，从而就操控了flow的流程停止。最终便是将Action的Abort状况传递给KisFlow的Abort状况。

已然有了Abort状况，那么咱们能够经过给Flow履行过程中增加一个设定，假如当时的Function没有提交本层的成果数据，也便是flow.buffer为空，那么将不会进入下一层，在本层直接完毕退出Flow的Run()调用。

kis-flow/flow/kis_flow_data.go

//commitCurData 提交Flow当时履行Function的成果数据
func (flow *KisFlow) commitCurData(ctx context.Context) error {
	// 判别本层核算是否有成果数据,假如没有则退出本次Flow Run循环
	if len(flow.buffer) == 0 {
        // ++++++++++++
		flow.abort = true
		return nil
	}
	// ... ...
    // ... ...
	return nil

7.1.5 Action捕获及处理

接下来来完成一个专门处理Action动作的办法，界说在kis-flow/flow/kis_flow_action.go文件中，如下：

kis-flow/flow/kis_flow_action.go

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) {
	if err := flow.commitCurData(ctx); err != nil {
		return nil, err
	}
	// 更新上一层 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
}

然后略微改善下KisFlow的Run() 流程，将dealAction() 办法嵌入进去。

kis-flow/flow/kis_flow.go

// Run 发动KisFlow的流式核算, 从开端Function开端履行流
func (flow *KisFlow) Run(ctx context.Context) error {
	var fn kis.Function
	fn = flow.FlowHead
	flow.abort = false
	if flow.Conf.Status == int(common.FlowDisable) {
		//flow被装备关闭
		return nil
	}
	// 由于此时还没有履行任何Function, 所以PrevFunctionId为FirstVirtual 由于没有上一层Function
	flow.PrevFunctionId = common.FunctionIdFirstVirtual
	// 提交数据流原始数据
	if err := flow.commitSrcData(ctx); err != nil {
		return err
	}
	//流式链式调用
	for fn != nil && flow.abort == false {
		// flow记录当时履行到的Function 符号
		fid := fn.GetId()
		flow.ThisFunction = fn
		flow.ThisFunctionId = fid
		// 得到当时Function要处理与的源数据
		if inputData, err := flow.getCurData(); err != nil {
			log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %sn", err.Error())
			return err
		} else {
			flow.inPut = inputData
		}
		if err := fn.Call(ctx, flow); err != nil {
			//Error
			return err
		} else {
			//Success
            // +++++++++++++++++++++++++++++++
			fn, err = flow.dealAction(ctx, fn)
			if err != nil {
				return err
			}
            // +++++++++++++++++++++++++++++++
		}
	}
	return nil
}

7.1.6 Action Abort单元测试

首要咱们新建一个Function事务，装备文件如下：

kis-flow/test/load_conf/func/func-AbortFunc.yml

kistype: func
fname: abortFunc
fmode: Calculate
source:
  name: 用户订单错误率
  must:
    - order_id
    - user_id

当时的Funciton的称号为abortFunc，然后完成其FaaS函数，如下：

kis-flow/test/faas/faas_abort.go

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)
}

这个Function就会终究调用flow.Next(kis.ActionAbort)来停止Flow，接下来咱们新建一个Flow，将上面的Function作为中心的Function，看检测是否会停止之后的Function被履行。
新建的flow的装备如下：

kis-flow/test/load_conf/flow/flow-FlowName2.yml

kistype: flow
status: 1
flow_name: flowName2
flows:
  - fname: funcName1
  - fname: abortFunc
  - fname: funcName3

当时Flow的称号为flowName2，当时的Flow有三个Function，其间funcNam1 和 funcName2咱们之前都现已界说好了，abortFunc是咱们新建的，而且在中心。假如abort功用满意，则funcName3将不会被调度。

接下来完成单元测试用例。

kis-flow/test/kis_action_test.go

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/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)
	}
}

其间下面的代码是初始化注册的代码，咱们也能够写在其他文件中，这样就不需求每次都带着这部分代码了。

	// 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)

cd 到kis-flow/test/目录下履行如下指令：

go test -test.v -test.paniconexit0 -test.run  TestActionAbort

成果如下：

=== RUN   TestActionAbort
Add KisPool FuncName=funcName1
Add KisPool FuncName=abortFunc
Add KisPool FuncName=funcName3
Add KisPool CaaSInit CName=ConnName1
Add KisPool CaaS CName=ConnName1, FName=funcName2, Mode =Save
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2]}
Add FlowRouter FlowName=flowName1
Add FlowRouter FlowName=flowName2
context.Background
====> After CommitSrcData, flow_name = flowName2, flow_id = flow-b6b90eb4b7d7457fbf85b3299b625513
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]
KisFunctionV, flow = &{Id:flow-b6b90eb4b7d7457fbf85b3299b625513 Name:flowName2 Conf:0xc000092cc0 Funcs:map[abortFunc:0xc000094d20 funcName1:0xc000094cc0 funcName3:0xc000094d80] FlowHead:0xc000094cc0 FlowTail:0xc000094d80 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000094cc0 ThisFunctionId:func-c435cf9f8e3346a1851f8c76375fce0f PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-7f5af1521fd64d08839d5bdd26de5254:map[] func-c435cf9f8e3346a1851f8c76375fce0f:map[] func-f0b80593fe2e4018a878f155b9c543b4:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] jumpFunc:NoJump abort:false nextOpt:<nil>}
---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-c435cf9f8e3346a1851f8c76375fce0f, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-c435cf9f8e3346a1851f8c76375fce0f, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-c435cf9f8e3346a1851f8c76375fce0f, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName2, flow_id = flow-b6b90eb4b7d7457fbf85b3299b625513
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-c435cf9f8e3346a1851f8c76375fce0f:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-b6b90eb4b7d7457fbf85b3299b625513 Name:flowName2 Conf:0xc000092cc0 Funcs:map[abortFunc:0xc000094d20 funcName1:0xc000094cc0 funcName3:0xc000094d80] FlowHead:0xc000094cc0 FlowTail:0xc000094d80 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000094d20 ThisFunctionId:func-7f5af1521fd64d08839d5bdd26de5254 PrevFunctionId:func-c435cf9f8e3346a1851f8c76375fce0f funcParams:map[func-7f5af1521fd64d08839d5bdd26de5254:map[] func-c435cf9f8e3346a1851f8c76375fce0f:map[] func-f0b80593fe2e4018a878f155b9c543b4:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-c435cf9f8e3346a1851f8c76375fce0f:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] jumpFunc:NoJump abort:false nextOpt:<nil>}
---> Call AbortFuncHandler ----
In FuncName = abortFunc, FuncId = func-7f5af1521fd64d08839d5bdd26de5254, row = data from funcName[funcName1], index = 0
In FuncName = abortFunc, FuncId = func-7f5af1521fd64d08839d5bdd26de5254, row = data from funcName[funcName1], index = 1
In FuncName = abortFunc, FuncId = func-7f5af1521fd64d08839d5bdd26de5254, row = data from funcName[funcName1], index = 2
--- PASS: TestActionAbort (0.00s)
PASS
ok      kis-flow/test   0.487s

经过成果能够看到，在履行完 AbortFuncHandler 后，没有持续履行，而是退出了Flow的Run()办法。

7.2 Action DataReuse（复用上层数据）

Action DataReuse 为服用上层数据，含义为，当时的履行Function提交到下一层的成果将不被运用，而是直接将当时Function的上一层成果数据，复用到下一层，作为下一层Funciton的数据源。

下面来完成Action DataReuse功用。

7.2.1 DataReuse Action增加

在Action中增加DataReuse成员，是一个bool类型。

kis-flow/kis/action.go

// Action KisFlow履行流程Actions
type Action struct {
    // +++++++++++++
	// DataReuse 是否复用上层Function数据
	DataReuse bool
	// Abort 停止Flow的履行
	Abort bool
}
// ActionDataReuse Next复用上层Function数据Option
func ActionDataReuse(act *Action) {
	act.DataReuse = true
}

然后提供一个ActionFunc，命名为：ActionDataReuse，完成中为改动DataReuse状况为true。

7.2.2 复用上层数据到基层

这儿需求再完成一个提交数据的办法，为怎么提交复用数据，详细逻辑如下：

kis-flow/flow/kis_flow_data.go

// commitReuseData
func (flow *KisFlow) commitReuseData(ctx context.Context) error {
	// 判别上层是否有成果数据, 假如没有则退出本次Flow Run循环
	if len(flow.data[flow.PrevFunctionId]) == 0 {
		flow.abort = true
		return nil
	}
	// 本层成果数据等于上层成果数据(复用上层成果数据到本层)
	flow.data[flow.ThisFunctionId] = flow.data[flow.PrevFunctionId]
	// 清空缓冲Buf (假如是ReuseData选项，那么提交的全部数据，都将不会带着到下一层)
	flow.buffer = flow.buffer[0:0]
	log.Logger().DebugFX(ctx, " ====> After commitReuseData, flow_name = %s, flow_id = %snAll Level Data =n %+vn", flow.Name, flow.Id, flow.data)
	return nil
}

逻辑很简单，与commitCurData()不同的是，commitCurData()为将flow.buffer的数据提交到flow.data[flow.ThisFunctionId]中，而commitReuseData()为将上一层的成果数据提交到flow.data[flow.ThisFunctionId]中。

7.2.3 处理DataReuse Action动作

然后在dealAction()办法中增加对Action DataReuse的动作捕获，如下：

kis-flow/flow/kis_flow_action.go

// 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
		}
	}
	// 更新上一层 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
}

7.2.4 单元测试

下面来针对DataReuse做单元测试，首要创立一个名字为dataReuseFunc 的Funciton，先创立装备文件。

kis-flow/test/load_conf/func/func-dataReuseFunc.yml

kistype: func
fname: dataReuseFunc
fmode: Calculate
source:
  name: 用户订单错误率
  must:
    - order_id
    - user_id

一起新建一个Flow流，称号为flowName3，装备如下：

kis-flow/test/load_conf/flow/func-FlowName3.yml

kistype: flow
status: 1
flow_name: flowName3
flows:
  - fname: funcName1
  - fname: dataReuseFunc
  - fname: funcName3

针对dataReuseFunc的Function的逻辑事务，如下：

kis-flow/test/faas/faas_data_reuse.go

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)
}

最终完成测试用例，如下：

kis-flow/test/kis_action_test.go

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)
	}
}

cd 到 kis-flow/test/下履行：

go test -test.v -test.paniconexit0 -test.run  TestActionDataReuse

成果是：

=== RUN   TestActionDataReuse
Add KisPool FuncName=funcName1
Add KisPool FuncName=dataReuseFunc
Add KisPool FuncName=funcName3
Add KisPool CaaSInit CName=ConnName1
Add KisPool CaaS CName=ConnName1, FName=funcName2, Mode =Save
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2]}
Add FlowRouter FlowName=flowName5
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2 funcName2]}
Add FlowRouter FlowName=flowName1
Add FlowRouter FlowName=flowName2
Add FlowRouter FlowName=flowName3
Add FlowRouter FlowName=flowName4
context.Background
====> After CommitSrcData, flow_name = flowName3, flow_id = flow-2c1a23d9587842bebaeee490319de81f
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]
KisFunctionV, flow = &{Id:flow-2c1a23d9587842bebaeee490319de81f Name:flowName3 Conf:0xc000092dc0 Funcs:map[dataReuseFunc:0xc000095620 funcName1:0xc0000955c0 funcName3:0xc000095680] FlowHead:0xc0000955c0 FlowTail:0xc000095680 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000955c0 ThisFunctionId:func-7886178381634f05b302841141382e59 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-7886178381634f05b302841141382e59:map[] func-cfe66e39aba54ff989d6764cc4edda20:map[] func-ef567879d0dd45b287ed709e549e9d32:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-7886178381634f05b302841141382e59, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-7886178381634f05b302841141382e59, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-7886178381634f05b302841141382e59, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName3, flow_id = flow-2c1a23d9587842bebaeee490319de81f
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-7886178381634f05b302841141382e59:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-2c1a23d9587842bebaeee490319de81f Name:flowName3 Conf:0xc000092dc0 Funcs:map[dataReuseFunc:0xc000095620 funcName1:0xc0000955c0 funcName3:0xc000095680] FlowHead:0xc0000955c0 FlowTail:0xc000095680 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000095620 ThisFunctionId:func-ef567879d0dd45b287ed709e549e9d32 PrevFunctionId:func-7886178381634f05b302841141382e59 funcParams:map[func-7886178381634f05b302841141382e59:map[] func-cfe66e39aba54ff989d6764cc4edda20:map[] func-ef567879d0dd45b287ed709e549e9d32:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-7886178381634f05b302841141382e59:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call DataReuseFuncHandler ----
In FuncName = dataReuseFunc, FuncId = func-ef567879d0dd45b287ed709e549e9d32, row = data from funcName[funcName1], index = 0
In FuncName = dataReuseFunc, FuncId = func-ef567879d0dd45b287ed709e549e9d32, row = data from funcName[funcName1], index = 1
In FuncName = dataReuseFunc, FuncId = func-ef567879d0dd45b287ed709e549e9d32, row = data from funcName[funcName1], index = 2
context.Background
 ====> After commitReuseData, flow_name = flowName3, flow_id = flow-2c1a23d9587842bebaeee490319de81f
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-7886178381634f05b302841141382e59:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] func-ef567879d0dd45b287ed709e549e9d32:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-2c1a23d9587842bebaeee490319de81f Name:flowName3 Conf:0xc000092dc0 Funcs:map[dataReuseFunc:0xc000095620 funcName1:0xc0000955c0 funcName3:0xc000095680] FlowHead:0xc0000955c0 FlowTail:0xc000095680 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000095680 ThisFunctionId:func-cfe66e39aba54ff989d6764cc4edda20 PrevFunctionId:func-ef567879d0dd45b287ed709e549e9d32 funcParams:map[func-7886178381634f05b302841141382e59:map[] func-cfe66e39aba54ff989d6764cc4edda20:map[] func-ef567879d0dd45b287ed709e549e9d32:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-7886178381634f05b302841141382e59:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] func-ef567879d0dd45b287ed709e549e9d32:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName3Handler ----
In FuncName = funcName3, FuncId = func-cfe66e39aba54ff989d6764cc4edda20, row = data from funcName[funcName1], index = 0
In FuncName = funcName3, FuncId = func-cfe66e39aba54ff989d6764cc4edda20, row = data from funcName[funcName1], index = 1
In FuncName = funcName3, FuncId = func-cfe66e39aba54ff989d6764cc4edda20, row = data from funcName[funcName1], index = 2
--- PASS: TestActionDataReuse (0.02s)
PASS
ok      kis-flow/test   0.523s

经过成果能够看出，在最终的funcName3Handler中得到的数据是funcName1传递下来的数据，中心的ReuseFunction将上层的数据复用到了下一层，变成了FuncName3的数据源。

7.3 Action ForceEntryNext（强制进入下一层）

7.3.1 ForceEntryNext Action特点

目前的KisFlow为，假如当时的Function没有commit数据（本层的成果数据），那么当时的Function完毕后，将不会持续调度下一层Function。可是有的Flow的流式核算或许需求持续向下履行，哪怕没有数据，所以这儿能够经过ForceEntryNext这个动作来触发。
首要咱们在Action中新增一个ForceEntryNext 特点。

kis-flow/kis/action.go

// Action KisFlow履行流程Actions
type Action struct {
	// DataReuse 是否复用上层Function数据
	DataReuse bool
	// 默许Next()为假如本层Function核算成果为0条数据，之后Function将不会持续履行
	// ForceEntryNext 为疏忽上述默许规则，没有数据强制进入下一层Function
	ForceEntryNext bool
	// Abort 停止Flow的履行
	Abort bool
}
// ActionForceEntryNext 强制进入下一层
func ActionForceEntryNext(act *Action) {
	act.ForceEntryNext = true
}

且提供装备函数ActionForceEntryNext()来修正这个特点状况。

7.3.2 捕获Action

在捕获Action的dealAction()办法中，加上对这个状况的判别，假如被设置，则需求将flow.Abort状况改成false，flow将持续履行下一层。

kis-flow/flow/kis_flow_action.go

// 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
	}
	// 更新上一层 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
}

这儿有一个细节，咱们需求调用一个办法commitVoidData()，即提交空数据，原因是，假如不提交空数据，那么flow.buffer依然为空，那么不会履行数据的提交动作，那么会导致flow.data[flow.ThisFunctionId]这条不存在，也便是key不存在，那么再履行到flow.getCurData()会呈现找不到key的异常而panic。所以这儿需求提交一个空的数据到flow.data[flow.ThisFunctionId]中。
详细的commitVoidData()完成如下：

kis-flow/flow/kis_flow_data.go

func (flow *KisFlow) commitVoidData(ctx context.Context) error {
	if len(flow.buffer) != 0 {
		return nil
	}
	// 制造空数据
	batch := make(common.KisRowArr, 0)
	// 将本层核算的缓冲数据提交到本层成果数据中
	flow.data[flow.ThisFunctionId] = batch
	log.Logger().DebugFX(ctx, " ====> After commitVoidData, flow_name = %s, flow_id = %snAll Level Data =n %+vn", flow.Name, flow.Id, flow.data)
	return nil
}

7.3.3 单元测试，不设置ForceEntryNext

首要，创立一个noResultFunc的Function装备，且完成相关的回调事务函数。

kis-flow/test/load_conf/func/func-NoResultFunc.yml

kistype: func
fname: noResultFunc
fmode: Calculate
source:
  name: 用户订单错误率
  must:
    - order_id
    - user_id

kis-flow/test/faas/faas_no_result.go

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()
}

这儿边在Function的最终，只调用flow.Next() 不传递任何Action动作。
然后新建一个FlowName4，装备如下：

kis-flow/test/load_conf/flow-FlowName4.yml

kistype: flow
status: 1
flow_name: flowName4
flows:
  - fname: funcName1
  - fname: noResultFunc
  - fname: funcName3

最终咱们编写单元测试用例代码，将noResultFunc放在中心的部分。

kis-flow/test/kis_action_test.go

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)
	}
}

cd到kis-flow/test/ 下履行：

go test -test.v -test.paniconexit0 -test.run  TestActionForceEntry

成果如下：

=== RUN   TestActionForceEntry
Add KisPool FuncName=funcName1
Add KisPool FuncName=noResultFunc
Add KisPool FuncName=funcName3
Add KisPool CaaSInit CName=ConnName1
Add KisPool CaaS CName=ConnName1, FName=funcName2, Mode =Save
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2]}
Add FlowRouter FlowName=flowName1
Add FlowRouter FlowName=flowName2
Add FlowRouter FlowName=flowName3
Add FlowRouter FlowName=flowName4
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2 funcName2]}
Add FlowRouter FlowName=flowName5
context.Background
====> After CommitSrcData, flow_name = flowName4, flow_id = flow-a496d02c79204e9a803fb5e1307523c9
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]
KisFunctionV, flow = &{Id:flow-a496d02c79204e9a803fb5e1307523c9 Name:flowName4 Conf:0xc000152e40 Funcs:map[funcName1:0xc00011d560 funcName3:0xc00011d620 noResultFunc:0xc00011d5c0] FlowHead:0xc00011d560 FlowTail:0xc00011d620 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc00011d560 ThisFunctionId:func-4d113d6a8e744d30a906db310f2d7818 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-47cb6f9ae464484aa779c18284035705:map[] func-4d113d6a8e744d30a906db310f2d7818:map[] func-70011c7ccecf46be91c6993d143639bb:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-4d113d6a8e744d30a906db310f2d7818, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-4d113d6a8e744d30a906db310f2d7818, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-4d113d6a8e744d30a906db310f2d7818, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName4, flow_id = flow-a496d02c79204e9a803fb5e1307523c9
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-4d113d6a8e744d30a906db310f2d7818:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-a496d02c79204e9a803fb5e1307523c9 Name:flowName4 Conf:0xc000152e40 Funcs:map[funcName1:0xc00011d560 funcName3:0xc00011d620 noResultFunc:0xc00011d5c0] FlowHead:0xc00011d560 FlowTail:0xc00011d620 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc00011d5c0 ThisFunctionId:func-47cb6f9ae464484aa779c18284035705 PrevFunctionId:func-4d113d6a8e744d30a906db310f2d7818 funcParams:map[func-47cb6f9ae464484aa779c18284035705:map[] func-4d113d6a8e744d30a906db310f2d7818:map[] func-70011c7ccecf46be91c6993d143639bb:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-4d113d6a8e744d30a906db310f2d7818:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call NoResultFuncHandler ----
In FuncName = noResultFunc, FuncId = func-47cb6f9ae464484aa779c18284035705, row = data from funcName[funcName1], index = 0
In FuncName = noResultFunc, FuncId = func-47cb6f9ae464484aa779c18284035705, row = data from funcName[funcName1], index = 1
In FuncName = noResultFunc, FuncId = func-47cb6f9ae464484aa779c18284035705, row = data from funcName[funcName1], index = 2
--- PASS: TestActionForceEntry (0.02s)
PASS
ok      kis-flow/test   0.958s

由于noResultFunc不会生成任何的成果数据，所以下一层Function将不会被履行，最终只履行到

---> Call NoResultFuncHandler ----

7.3.4 单元测试，设置ForceEntryNext

下面咱们将Action为ForceEntryNext加上，在NoResultFuncHandler() 中，加上flow.Next(kis.ActionForceEntryNext)，如下：

kis-flow/test/faas/faas_no_result.go

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)
}

cd到kis-flow/test/ 下履行：

go test -test.v -test.paniconexit0 -test.run  TestActionForceEntry

成果如下：

=== RUN   TestActionForceEntry
Add KisPool FuncName=funcName1
Add KisPool FuncName=noResultFunc
Add KisPool FuncName=funcName3
Add KisPool CaaSInit CName=ConnName1
Add KisPool CaaS CName=ConnName1, FName=funcName2, Mode =Save
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2]}
Add FlowRouter FlowName=flowName5
===> Call Connector InitDemo1
&{conn ConnName1 0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990 redis redis-key map[args1:value1 args2:value2] [] [funcName2 funcName2 funcName2]}
Add FlowRouter FlowName=flowName1
Add FlowRouter FlowName=flowName2
Add FlowRouter FlowName=flowName3
Add FlowRouter FlowName=flowName4
context.Background
====> After CommitSrcData, flow_name = flowName4, flow_id = flow-7fb47f227c9f4b9d8fa69c28177fc7bb
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]
KisFunctionV, flow = &{Id:flow-7fb47f227c9f4b9d8fa69c28177fc7bb Name:flowName4 Conf:0xc000028e80 Funcs:map[funcName1:0xc0000136e0 funcName3:0xc0000137a0 noResultFunc:0xc000013740] FlowHead:0xc0000136e0 FlowTail:0xc0000137a0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000136e0 ThisFunctionId:func-ecddaee7d7d447a9852d07088732f509 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-5729600ae6ea4d6f879eb5832c638e1a:map[] func-c9817c7993894919b8463dea1757544e:map[] func-ecddaee7d7d447a9852d07088732f509:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-ecddaee7d7d447a9852d07088732f509, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-ecddaee7d7d447a9852d07088732f509, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-ecddaee7d7d447a9852d07088732f509, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName4, flow_id = flow-7fb47f227c9f4b9d8fa69c28177fc7bb
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-ecddaee7d7d447a9852d07088732f509:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-7fb47f227c9f4b9d8fa69c28177fc7bb Name:flowName4 Conf:0xc000028e80 Funcs:map[funcName1:0xc0000136e0 funcName3:0xc0000137a0 noResultFunc:0xc000013740] FlowHead:0xc0000136e0 FlowTail:0xc0000137a0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000013740 ThisFunctionId:func-c9817c7993894919b8463dea1757544e PrevFunctionId:func-ecddaee7d7d447a9852d07088732f509 funcParams:map[func-5729600ae6ea4d6f879eb5832c638e1a:map[] func-c9817c7993894919b8463dea1757544e:map[] func-ecddaee7d7d447a9852d07088732f509:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-ecddaee7d7d447a9852d07088732f509:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call NoResultFuncHandler ----
In FuncName = noResultFunc, FuncId = func-c9817c7993894919b8463dea1757544e, row = data from funcName[funcName1], index = 0
In FuncName = noResultFunc, FuncId = func-c9817c7993894919b8463dea1757544e, row = data from funcName[funcName1], index = 1
In FuncName = noResultFunc, FuncId = func-c9817c7993894919b8463dea1757544e, row = data from funcName[funcName1], index = 2
context.Background
 ====> After commitVoidData, flow_name = flowName4, flow_id = flow-7fb47f227c9f4b9d8fa69c28177fc7bb
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-c9817c7993894919b8463dea1757544e:[] func-ecddaee7d7d447a9852d07088732f509:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-7fb47f227c9f4b9d8fa69c28177fc7bb Name:flowName4 Conf:0xc000028e80 Funcs:map[funcName1:0xc0000136e0 funcName3:0xc0000137a0 noResultFunc:0xc000013740] FlowHead:0xc0000136e0 FlowTail:0xc0000137a0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000137a0 ThisFunctionId:func-5729600ae6ea4d6f879eb5832c638e1a PrevFunctionId:func-c9817c7993894919b8463dea1757544e funcParams:map[func-5729600ae6ea4d6f879eb5832c638e1a:map[] func-c9817c7993894919b8463dea1757544e:map[] func-ecddaee7d7d447a9852d07088732f509:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-c9817c7993894919b8463dea1757544e:[] func-ecddaee7d7d447a9852d07088732f509:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName3Handler ----
--- PASS: TestActionForceEntry (0.01s)
PASS
ok      kis-flow/test   0.348s

会发现，Function第三层funcName3Handler 被履行到，可是没有任何的数据。

7.4 Action JumpFunc（流程跳转）

接下来，来完成JumpFunc Action，JumpFunc是能够在当时Flow中恣意跳转到指定的FuncName持续履行（前提是跳转的FuncName当当时Flow中存在）

留意：JumpFunc简单呈现无限循环流，所以在事务的规划要慎用。

7.4.1 Action增加JumpFunc

首要在Action增加一个JumpFunc特点，留意，JunpFunc不是一个bool状况，而是一个string字符串，表明详细要跳转的FunctionName称号。

kis-flow/kis/action.go

// 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
}
// ActionJumpFunc 会回来一个ActionFunc函数，而且会将funcName赋值给Action.JumpFunc
// (留意：简单呈现Flow循环调用，导致死循环)
func ActionJumpFunc(funcName string) ActionFunc {
	return func(act *Action) {
		act.JumpFunc = funcName
	}
}

然后提供一个修正JumpFunc的装备办法ActionJumpFunc(),留意这个办法和之前的办法写法有一些不同，主要是回来一个匿名函数而且履行，意图则是修正Action中的JumpFunc特点。

7.4.2 捕获Action

接下来，咱们来捕获JumpFunc的Action动作，判别JumpFunc是否为空字符串即可。

kis-flow/flow/kis_flow_action.go

// 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
}

假如设置JumpFunc，则需求修正下次履行的fn指针，否则则正常寻址fn.Next()。

7.4.3 单元测试

接下来来界说一个跳转Action的Function，装备，如下：

kis-flow/test/load_conf/func/func-jumpFunc.yml

kistype: func
fname: jumpFunc
fmode: Calculate
source:
  name: 用户订单错误率
  must:
    - order_id
    - user_id

而且完成相关的Funciton事务逻辑，如下：

kis-flow/test/faas/faas_jump.go

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"))
}

这儿，最终经过flow.Next(kis.ActionJumpFunc("funcName1"))来指定跳转到funcName1的Function。

新建一个Flow，为FlowName5，装备如下：

kis-flow/test/load_conf/flow/flow-FlowName5.yml

kistype: flow
status: 1
flow_name: flowName5
flows:
  - fname: funcName1
  - fname: funcName2
  - fname: jumpFunc

之后，来完成单元测试用例代码，如下：

kis-flow/test/kis_action_test.go

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)
	}
}

cd到kis-flow/test/履行：

go test -test.v -test.paniconexit0 -test.run  TestActionJumpFunc

成果如下：

...
...
---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-f6ca8010d66744429bf6069c9897a928, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-f6ca8010d66744429bf6069c9897a928, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-f6ca8010d66744429bf6069c9897a928, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName5, flow_id = flow-5da80af989dc49648a001762fa08b866
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-5800567c4cd842b6b377c2b0c0fd81c2:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f6ca8010d66744429bf6069c9897a928:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionS, flow = &{Id:flow-5da80af989dc49648a001762fa08b866 Name:flowName5 Conf:0xc000028f80 Funcs:map[funcName1:0xc000013620 funcName2:0xc000013680 jumpFunc:0xc0000136e0] FlowHead:0xc000013620 FlowTail:0xc0000136e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000013680 ThisFunctionId:func-5800567c4cd842b6b377c2b0c0fd81c2 PrevFunctionId:func-f6ca8010d66744429bf6069c9897a928 funcParams:map[func-4faf8f019f4a4a48b84ef27abfad53d1:map[] func-5800567c4cd842b6b377c2b0c0fd81c2:map[] func-f6ca8010d66744429bf6069c9897a928:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-5800567c4cd842b6b377c2b0c0fd81c2:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f6ca8010d66744429bf6069c9897a928:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName2Handler ----
In FuncName = funcName2, FuncId = func-5800567c4cd842b6b377c2b0c0fd81c2, row = data from funcName[funcName1], index = 0
===> In CaasDemoHanler1: flowName: flowName5, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 0
In FuncName = funcName2, FuncId = func-5800567c4cd842b6b377c2b0c0fd81c2, row = data from funcName[funcName1], index = 1
===> In CaasDemoHanler1: flowName: flowName5, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 1
In FuncName = funcName2, FuncId = func-5800567c4cd842b6b377c2b0c0fd81c2, row = data from funcName[funcName1], index = 2
===> In CaasDemoHanler1: flowName: flowName5, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 2
context.Background
 ====> After commitCurData, flow_name = flowName5, flow_id = flow-5da80af989dc49648a001762fa08b866
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-5800567c4cd842b6b377c2b0c0fd81c2:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f6ca8010d66744429bf6069c9897a928:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]
KisFunctionC, flow = &{Id:flow-5da80af989dc49648a001762fa08b866 Name:flowName5 Conf:0xc000028f80 Funcs:map[funcName1:0xc000013620 funcName2:0xc000013680 jumpFunc:0xc0000136e0] FlowHead:0xc000013620 FlowTail:0xc0000136e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000136e0 ThisFunctionId:func-4faf8f019f4a4a48b84ef27abfad53d1 PrevFunctionId:func-5800567c4cd842b6b377c2b0c0fd81c2 funcParams:map[func-4faf8f019f4a4a48b84ef27abfad53d1:map[] func-5800567c4cd842b6b377c2b0c0fd81c2:map[] func-f6ca8010d66744429bf6069c9897a928:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-5800567c4cd842b6b377c2b0c0fd81c2:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f6ca8010d66744429bf6069c9897a928:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call JumpFuncHandler ----
In FuncName = jumpFunc, FuncId = func-4faf8f019f4a4a48b84ef27abfad53d1, row = data from funcName[funcName2], index = 0
In FuncName = jumpFunc, FuncId = func-4faf8f019f4a4a48b84ef27abfad53d1, row = data from funcName[funcName2], index = 1
In FuncName = jumpFunc, FuncId = func-4faf8f019f4a4a48b84ef27abfad53d1, row = data from funcName[funcName2], index = 2
KisFunctionV, flow = &{Id:flow-5da80af989dc49648a001762fa08b866 Name:flowName5 Conf:0xc000028f80 Funcs:map[funcName1:0xc000013620 funcName2:0xc000013680 jumpFunc:0xc0000136e0] FlowHead:0xc000013620 FlowTail:0xc0000136e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000013620 ThisFunctionId:func-f6ca8010d66744429bf6069c9897a928 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-4faf8f019f4a4a48b84ef27abfad53d1:map[] func-5800567c4cd842b6b377c2b0c0fd81c2:map[] func-f6ca8010d66744429bf6069c9897a928:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-5800567c4cd842b6b377c2b0c0fd81c2:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f6ca8010d66744429bf6069c9897a928:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] abort:false action:{DataReuse:false ForceEntryNext:false JumpFunc: Abort:false}}
---> Call funcName1Handler ----
 ... 
 ...

发现咱们会无限循环的调度Flow，这样阐明咱们的JumpFunc Action现已生效。

7.5【V0.6】源代码

github.com/aceld/kis-f…

作者:刘丹冰Aceld github: github.com/aceld

KisFlow开源项目地址：github.com/aceld/kis-f…

Golang结构实战-KisFlow流式核算结构专栏

Golang结构实战-KisFlow流式核算结构(1)-概述

Golang结构实战-KisFlow流式核算结构(2)-项目构建/根底模块-(上)

Golang结构实战-KisFlow流式核算结构(3)-项目构建/根底模块-(下)

Golang结构实战-KisFlow流式核算结构(4)-数据流

Golang结构实战-KisFlow流式核算结构(5)-Function调度

Golang结构实战-KisFlow流式核算结构(6)-Connector

Golang框架实战-KisFlow流式计算框架(8)-KisFlow Action

Golang结构实战-KisFlow流式核算结构专栏

7.1 Action Abort（停止流程）

7.1.1 Abort接口界说

7.1.2 Action模块界说

7.1.3 Next办法完成

7.1.4 Abort操控Flow流程

7.1.5 Action捕获及处理

7.1.6 Action Abort单元测试

7.2 Action DataReuse（复用上层数据）

7.2.1 DataReuse Action增加

7.2.2 复用上层数据到基层

7.2.3 处理DataReuse Action动作

7.2.4 单元测试

7.3 Action ForceEntryNext（强制进入下一层）

7.3.1 ForceEntryNext Action特点

7.3.2 捕获Action

7.3.3 单元测试，不设置ForceEntryNext

7.3.4 单元测试，设置ForceEntryNext

7.4 Action JumpFunc（流程跳转）

7.4.1 Action增加JumpFunc

7.4.2 捕获Action

7.4.3 单元测试

7.5【V0.6】源代码

Golang结构实战-KisFlow流式核算结构专栏

相关文章

Docker 有什么优点？

黑炫酷的监控界面，实际上是用了什么开源工具？

Java线程安全的集合类ConcurrentHashMap详解

规则引擎调研及初步使用 | 京东云技术团队

作者信息