(Part 6)Golang Framework Hands-on - KisFlow Stream Computing Framework-Configuration Import and Export
Tags: go
Published: 2024-05-13 07:35:50

<img width="150px" src="https://github.com/aceld/kis-flow/assets/7778936/8729d750-897c-4ba3-98b4-c346188d034e" />

Github: https://github.com/aceld/kis-flow
Document: https://github.com/aceld/kis-flow/wiki

---

[Part1-OverView](https://dev.to/aceld/part-1-golang-framework-hands-on-kisflow-streaming-computing-framework-overview-8fh)
[Part2.1-Project Construction / Basic Modules](https://dev.to/aceld/part-2-golang-framework-hands-on-kisflow-streaming-computing-framework-project-construction-basic-modules-cia)
[Part2.2-Project Construction / Basic Modules](https://dev.to/aceld/part-3golang-framework-hands-on-kisflow-stream-computing-framework-project-construction-basic-modules-1epb)
[Part3-Data Stream](https://dev.to/aceld/part-4golang-framework-hands-on-kisflow-stream-computing-framework-data-stream-1mbd)
[Part4-Function Scheduling](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-function-scheduling-4p0h)
[Part5-Connector](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-connector-hcd)
[Part6-Configuration Import and Export](https://dev.to/aceld/part-6golang-framework-hands-on-kisflow-stream-computing-framework-configuration-import-and-export-47o1)
[Part7-KisFlow Action](https://dev.to/aceld/part-7golang-framework-hands-on-kisflow-stream-computing-framework-kisflow-action-3n05)
[Part8-Cache/Params Data Caching and Data Parameters](https://dev.to/aceld/part-8golang-framework-hands-on-cacheparams-data-caching-and-data-parameters-5df5)
[Part9-Multiple Copies of Flow](https://dev.to/aceld/part-8golang-framework-hands-on-multiple-copies-of-flow-c4k)
[Part10-Prometheus Metrics Statistics](https://dev.to/aceld/part-10golang-framework-hands-on-prometheus-metrics-statistics-22f0)
[Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection](https://dev.to/aceld/part-11golang-framework-hands-on-adaptive-registration-of-faas-parameter-types-based-on-reflection-15i9)


---

[Case1-Quick Start](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-quick-start-guide-f51)
[Case2-Flow Parallel Operation](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-flow-parallel-operation-364m)


---

## 6.1 Configuration Import

Currently, establishing a Flow and Functions requires a series of cumbersome additions each time, which is not very convenient. Next, we can construct the structure of KisFlow and also export the structure of KisFlow to local files by batch reading and writing configuration files. Currently, we will use file format for configuration persistence, but developers can also choose to do persistence with databases or remote configurations in the future.

### 6.1.1 Creating Configuration Files

First, we create the necessary configuration files for KisFlow business under `kis-flow/test/load_conf/`. Within `kis-flow/test/load_conf/`, we create three folders: `conn/`, `flow/`, and `func/` to respectively store the configuration information for `Connector`, `Flow`, and `Function`.

```bash
├── conn
│   └── conn-ConnName1.yml
├── flow
│   └── flow-FlowName1.yml
└── func
    ├── func-FuncName1.yml
    ├── func-FuncName2.yml
    └── func-FuncName3.yml
```

Create some `yml` files within these folders. The specific content is as follows:

#### A. Function

> kis-flow/test/load_conf/func/func-FuncNam1.yml
```yaml
kistype: func
fname: funcName1
fmode: Verify
source:
  name: Public account Douyin mall user order data
  must:
    - order_id
    - user_id
```

> kis-flow/test/load_conf/func/func-FuncNam2.yml
```yaml
kistype: func
fname: funcName2
fmode: Save
source:
  name: User order error rate
  must:
    - order_id
    - user_id
option:
  cname: ConnName1
```

> kis-flow/test/load_conf/func/func-FuncNam3.yml

```yaml
kistype: func
fname: funcName3
fmode: Calculate
source:
  name: User order error rate
  must:
    - order_id
    - user_id
```

#### B. Connector

> kis-flow/test/load_conf/func/func-ConnName1.yml

```yaml
kistype: conn
cname: ConnName1
addrs: '0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990'
type: redis
key: redis-key
params:
  args1: value1
  args2: value2
load: null
save:
  - funcName2
```

#### C. Flow
> kis-flow/test/load_conf/func/func-FlowName1.yml

```yaml
kistype: flow
status: 1
flow_name: flowName1
flows:
  - fname: funcName1
  - fname: funcName2
  - fname: funcName3

```

### 6.1.2 Configuration Parsing

Create the `kis-flow/file/` directory and the `kis-flow/file/config_import.go` file within it.

First, define an interface allConfig that can hold all configurations:

> kis-flow/file/config_import.go

```go
type allConfig struct {
	Flows map[string]*config.KisFlowConfig
	Funcs map[string]*config.KisFuncConfig
	Conns map[string]*config.KisConnConfig
}
```

Where the key serves as the Name field for each module.
Next, define methods for parsing Flow, Function, and Connector configurations. For YAML parsing, we will use the "gopkg.in/yaml.v3" library.

#### A. Flow Configuration Parsing

> kis-flow/file/config_import.go

```go
// kisTypeFlowConfigure parses Flow configuration files in YAML format
func kisTypeFlowConfigure(all *allConfig, confData []byte, fileName string, kisType interface{}) error {
	flow := new(config.KisFlowConfig)
	if ok := yaml.Unmarshal(confData, flow); ok != nil {
		return errors.New(fmt.Sprintf("%s has wrong format kisType = %s", fileName, kisType))
	}

	// Do not load Flow configuration if the status is disabled
	if common.KisOnOff(flow.Status) == common.FlowDisable {
		return nil
	}

	if _, ok := all.Flows[flow.FlowName]; ok {
		return errors.New(fmt.Sprintf("%s set repeat flow_id:%s", fileName, flow.FlowName))
	}

	// Add the configuration to the collection
	all.Flows[flow.FlowName] = flow

	return nil
}
```

* `confData`: the file's binary data
* `fileName`: the file path
* `kistype`: the configuration file type

`kisTypeFlowConfigure` will parse the configuration information into the `Flows` member of allConfig.
Similarly, the methods for `Function` and `Connector` parsing are as follows.

#### B. Function Configuration Parsing

> kis-flow/file/config_import.go

```go
// kisTypeFuncConfigure parses Function configuration files in YAML format
func kisTypeFuncConfigure(all *allConfig, confData []byte, fileName string, kisType interface{}) error {
	function := new(config.KisFuncConfig)
	if ok := yaml.Unmarshal(confData, function); ok != nil {
		return errors.New(fmt.Sprintf("%s has wrong format kisType = %s", fileName, kisType))
	}
	if _, ok := all.Funcs[function.FName]; ok {
		return errors.New(fmt.Sprintf("%s set repeat function_id:%s", fileName, function.FName))
	}

	// Add the configuration to the collection
	all.Funcs[function.FName] = function

	return nil
}
```

#### C. Connector Configuration Parsing
> kis-flow/file/config_import.go

```go
// kisTypeConnConfigure parses Connector configuration files in YAML format
func kisTypeConnConfigure(all *allConfig, confData []byte, fileName string, kisType interface{}) error {
	conn := new(config.KisConnConfig)
	if ok := yaml.Unmarshal(confData, conn); ok != nil {
		return errors.New(fmt.Sprintf("%s is wrong format nsType = %s", fileName, kisType))
	}

	if _, ok := all.Conns[conn.CName]; ok {
		return errors.New(fmt.Sprintf("%s set repeat conn_id:%s", fileName, conn.CName))
	}

	// Add the configuration to the collection
	all.Conns[conn.CName] = conn

	return nil
}

```

### 6.1.3 Traversing Files

Below is an implementation to traverse all `yml` and `yaml` type files under a given path loadPath and parse the configuration information into `allConfig` according to the kistype category.

> kis-flow/file/config_import.go

```go
// parseConfigWalkYaml parses all configuration files in YAML format and stores the configuration information in allConfig
func parseConfigWalkYaml(loadPath string) (*allConfig, error) {

	all := new(allConfig)

	all.Flows = make(map[string]*config.KisFlowConfig)
	all.Funcs = make(map[string]*config.KisFuncConfig)
	all.Conns = make(map[string]*config.KisConnConfig)

	err := filepath.Walk(loadPath, func(filePath string, info os.FileInfo, err error) error {
		// Validate file extension
		if suffix := path.Ext(filePath); suffix != ".yml" && suffix != ".yaml" {
			return nil
		}

		// Read file content
		confData, err := ioutil.ReadFile(filePath)
		if err != nil {
			return err
		}

		confMap := make(map[string]interface{})

		// Validate YAML format
		if err := yaml.Unmarshal(confData, confMap); err != nil {
			return err
		}

		// Check if kisType exists
		if kisType, ok := confMap["kistype"]; !ok {
			return errors.New(fmt.Sprintf("yaml file %s has no file [kistype]!", filePath))
		} else {
			switch kisType {
			case common.KisIdTypeFlow:
				return kisTypeFlowConfigure(all, confData, filePath, kisType)

			case common.KisIdTypeFunction:
				return kisTypeFuncConfigure(all, confData, filePath, kisType)

			case common.KisIdTypeConnnector:
				return kisTypeConnConfigure(all, confData, filePath, kisType)

			default:
				return errors.New(fmt.Sprintf("%s set wrong kistype %s", filePath, kisType))
			}
		}
	})

	if err != nil {
		return nil, err
	}

	return all, nil
}

```

### 6.1.4 Import Method

Below is a public method `ConfigImportYaml` that imports files from a given root path.

> kis-flow/file/config_import.go

```go
// ConfigImportYaml parses all configuration files in YAML format
func ConfigImportYaml(loadPath string) error {

	all, err := parseConfigWalkYaml(loadPath)
	if err != nil {
		return err
	}

	for flowName, flowConfig := range all.Flows {

		// Build a Flow
		newFlow := flow.NewKisFlow(flowConfig)

		for _, fp := range flowConfig.Flows {
			if err := buildFlow(all, fp, newFlow, flowName); err != nil {
				return err
			}
		}

		// Add the flow to the FlowPool
		kis.Pool().AddFlow(flowName, newFlow)
	}

	return nil
}
```

First, it calls `parseConfigWalkYaml()` to load all configuration information into memory.
Then, it traverses all Flows one by one to construct them. Finally, the flow is added to the Pool. The specific construction process is as follows:

> kis-flow/file/config_import.go

```go
func buildFlow(all *allConfig, fp config.KisFlowFunctionParam, newFlow kis.Flow, flowName string) error {
	// Load Functions that the current Flow depends on
	if funcConfig, ok := all.Funcs[fp.FuncName]; !ok {
		return errors.New(fmt.Sprintf("FlowName [%s] need FuncName [%s], But has No This FuncName Config", flowName, fp.FuncName))
	} else {
		// Link Flow to Functions
		if funcConfig.Option.CName != "" {
			// Load Connectors that the current Function depends on
			if connConf, ok := all.Conns[funcConfig.Option.CName]; !ok {
				return errors.New(fmt.Sprintf("FuncName [%s] need ConnName [%s], But has No This ConnName Config", fp.FuncName, funcConfig.Option.CName))
			} else {
				// Associate Function Configurations with Connector Configurations
				_ = funcConfig.AddConnConfig(connConf)
			}
		}

		// Link Flow to Functions
		if err := newFlow.Link(funcConfig, fp.Params); err != nil {
			return err
		}
	}

	return nil
}
```

## 6.2 Configure Import Unit Testing

> Create a unit test file kis-flow/test/kis_config_import_test.go.

```go
package test

import (
	"context"
	"kis-flow/common"
	"kis-flow/file"
	"kis-flow/kis"
	"kis-flow/test/caas"
	"kis-flow/test/faas"
	"testing"
)

func TestConfigImportYmal(t *testing.T) {
	ctx := context.Background()

	// 0. Register Function callbacks
	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
	kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler)
	kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)

	// 0. Register ConnectorInit and Connector callbacks
	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)

	// 1. Load configuration file and build Flow
	if err := file.ConfigImportYaml("/Users/gopath/src/kis-flow/test/load_conf/"); err != nil {
		panic(err)
	}

	// 2. Get Flow
	flow1 := kis.Pool().GetFlow("flowName1")

	// 3. Commit raw data
	_ = flow1.CommitRow("This is Data1 from Test")
	_ = flow1.CommitRow("This is Data2 from Test")
	_ = flow1.CommitRow("This is Data3 from Test")

	// 4. Execute flow1
	if err := flow1.Run(ctx); err != nil {
		panic(err)
	}
}

```

First, register the business methods. Then load the configuration via ConfigImportYaml, get the flow instance from the pool, submit data, and run it.

> Note that the absolute path to the configuration file is provided here.

Navigate to the `kis-flow/test/` directory and execute the command:

```bash
go test -test.v -test.paniconexit0 -test.run TestConfigImportYmal    
```

The expected result is as follows:

```bash
=== RUN   TestConfigImportYmal
Add KisPool FuncName=funcName1
Add KisPool FuncName=funcName2
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

context.Background
====> After CommitSrcData, flow_name = flowName1, flow_id = flow-bcaaa02a8d4b4a80b2f2895d9cecf20b
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]

KisFunctionV, flow = &{Id:flow-bcaaa02a8d4b4a80b2f2895d9cecf20b Name:flowName1 Conf:0xc00014eb00 Funcs:map[funcName1:0xc000114960 funcName2:0xc0001149c0 funcName3:0xc000114a20] FlowHead:0xc000114960 FlowTail:0xc000114a20 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000114960 ThisFunctionId:func-37c7070f45144529891d433ae9c4ebfc PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-37c7070f45144529891d433ae9c4ebfc:map[] func-5315301ffbbb4ae4be021729ddff1569:map[] func-89a6a662729b4a0895e849c40bf29892: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]}

---> Call funcName1Handler ----
In FuncName = funcName1, FuncId = func-37c7070f45144529891d433ae9c4ebfc, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-37c7070f45144529891d433ae9c4ebfc, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-37c7070f45144529891d433ae9c4ebfc, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName1, flow_id = flow-bcaaa02a8d4b4a80b2f2895d9cecf20b
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-37c7070f45144529891d433ae9c4ebfc:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]

KisFunctionS, flow = &{Id:flow-bcaaa02a8d4b4a80b2f2895d9cecf20b Name:flowName1 Conf:0xc00014eb00 Funcs:map[funcName1:0xc000114960 funcName2:0xc0001149c0 funcName3:0xc000114a20] FlowHead:0xc000114960 FlowTail:0xc000114a20 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0001149c0 ThisFunctionId:func-5315301ffbbb4ae4be021729ddff1569 PrevFunctionId:func-37c7070f45144529891d433ae9c4ebfc funcParams:map[func-37c7070f45144529891d433ae9c4ebfc:map[] func-5315301ffbbb4ae4be021729ddff1569:map[] func-89a6a662729b4a0895e849c40bf29892: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-37c7070f45144529891d433ae9c4ebfc:[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]}

---> Call funcName2Handler ----
In FuncName = funcName2, FuncId = func-5315301ffbbb4ae4be021729ddff1569, row = data from funcName[funcName1], index = 0
===> In CaasDemoHanler1: flowName: flowName1, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 0
In FuncName = funcName2, FuncId = func-5315301ffbbb4ae4be021729ddff1569, row = data from funcName[funcName1], index = 1
===> In CaasDemoHanler1: flowName: flowName1, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 1
In FuncName = funcName2, FuncId = func-5315301ffbbb4ae4be021729ddff1569, row = data from funcName[funcName1], index = 2
===> In CaasDemoHanler1: flowName: flowName1, cName:ConnName1, fnName:funcName2, mode:Save
===> Call Connector CaasDemoHanler1, args from funciton: data from funcName[funcName1], index = 2
context.Background
 ====> After commitCurData, flow_name = flowName1, flow_id = flow-bcaaa02a8d4b4a80b2f2895d9cecf20b
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-37c7070f45144529891d433ae9c4ebfc:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] func-5315301ffbbb4ae4be021729ddff1569:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2]]

KisFunctionC, flow = &{Id:flow-bcaaa02a8d4b4a80b2f2895d9cecf20b Name:flowName1 Conf:0xc00014eb00 Funcs:map[funcName1:0xc000114960 funcName2:0xc0001149c0 funcName3:0xc000114a20] FlowHead:0xc000114960 FlowTail:0xc000114a20 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000114a20 ThisFunctionId:func-89a6a662729b4a0895e849c40bf29892 PrevFunctionId:func-5315301ffbbb4ae4be021729ddff1569 funcParams:map[func-37c7070f45144529891d433ae9c4ebfc:map[] func-5315301ffbbb4ae4be021729ddff1569:map[] func-89a6a662729b4a0895e849c40bf29892: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-37c7070f45144529891d433ae9c4ebfc:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2] func-5315301ffbbb4ae4be021729ddff1569:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2]] inPut:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2]}

---> Call funcName3Handler ----
In FuncName = funcName3, FuncId = func-89a6a662729b4a0895e849c40bf29892, row = data from funcName[funcName2], index = 0
In FuncName = funcName3, FuncId = func-89a6a662729b4a0895e849c40bf29892, row = data from funcName[funcName2], index = 1
In FuncName = funcName3, FuncId = func-89a6a662729b4a0895e849c40bf29892, row = data from funcName[funcName2], index = 2
--- PASS: TestConfigImportYmal (0.01s)
PASS
ok      kis-flow/test   0.517s

```

The expected result is consistent with our expectations. Now we can load and build KisFlow through configuration files.



## 6.3 Configure Export

### 6.3.1 Implementation of Export

> kis-flow/file/config_export.go

```go
package file

import (
	"errors"
	"fmt"
	"gopkg.in/yaml.v3"
	"io/ioutil"
	"kis-flow/common"
	"kis-flow/kis"
)

// ConfigExportYaml exports flow configurations and stores them locally
func ConfigExportYaml(flow kis.Flow, savePath string) error {

	if data, err := yaml.Marshal(flow.GetConfig()); err != nil {
		return err
	} else {
		// Flow
		err := ioutil.WriteFile(savePath+common.KisIdTypeFlow+"-"+flow.GetName()+".yaml", data, 0644)
		if err != nil {
			return err
		}

		// Function
		for _, fp := range flow.GetConfig().Flows {
			fConf := flow.GetFuncConfigByName(fp.FuncName)
			if fConf == nil {
				return errors.New(fmt.Sprintf("function name = %s config is nil ", fp.FuncName))
			}

			if fdata, err := yaml.Marshal(fConf); err != nil {
				return err
			} else {
				if err := ioutil.WriteFile(savePath+common.KisIdTypeFunction+"-"+fp.FuncName+".yaml", fdata, 0644); err != nil {
					return err
				}
			}

			// Connector
			if fConf.Option.CName != "" {
				cConf, err := fConf.GetConnConfig()
				if err != nil {
					return err
				}
				if cdata, err := yaml.Marshal(cConf); err != nil {
					return err
				} else {
					if err := ioutil.WriteFile(savePath+common.KisIdTypeConnnector+"-"+cConf.CName+".yaml", cdata, 0644); err != nil {
						return err
					}
				}
			}
		}
	}

	return nil
}
```


### 6.3.2 New Interfaces in Flow
> kis-flow/kis/flow.go
```go
package kis

import (
	"context"
	"kis-flow/common"
	"kis-flow/config"
)

type Flow interface {
	// Run schedules the Flow, sequentially dispatches and executes Functions in the Flow
	Run(ctx context.Context) error
	// Link connects the Functions in the Flow according to the configuration file
	Link(fConf *config.KisFuncConfig, fParams config.FParam) error
	// CommitRow submits Flow data to the upcoming Function layer
	CommitRow(row interface{}) error
	// Input gets the input source data for the currently executing Function of the flow
	Input() common.KisRowArr
	// GetName gets the name of the Flow
	GetName() string
	// GetThisFunction gets the currently executing Function of the Flow
	GetThisFunction() Function
	// GetThisFuncConf gets the configuration of the currently executing Function of the Flow
	GetThisFuncConf() *config.KisFuncConfig
	// GetConnector gets the Connector of the currently executing Function of the Flow
	GetConnector() (Connector, error)
	// GetConnConf gets the configuration of the Connector of the currently executing Function of the Flow
	GetConnConf() (*config.KisConnConfig, error)
    
    // +++++++++++++++++++++++++++++++
	// GetConfig gets the configuration of the current Flow
	GetConfig() *config.KisFlowConfig
	// GetFuncConfigByName gets the configuration of the current Flow by Function name
	GetFuncConfigByName(funcName string) *config.KisFuncConfig
    // +++++++++++++++++++++++++++++++
}
```

Flow's new interfaces are implemented as follows:

> kis-flow/flow/kis_flow.go

```go
func (flow *KisFlow) GetConfig() *config.KisFlowConfig {
	return flow.Conf
}

// GetFuncConfigByName gets the configuration of the current Flow by Function name
func (flow *KisFlow) GetFuncConfigByName(funcName string) *config.KisFuncConfig {
	if f, ok := flow.Funcs[funcName]; ok {
		return f.GetConfig()
	} else {
		log.Logger().ErrorF("GetFuncConfigByName(): Function %s not found", funcName)
		return nil
	}
}
```

### 6.3.3 Fix in KisFlow's Funcs
There was a typo here before.

> kis-flow/flow/kis_flow.go
```go
// KisFlow is used to pass the context environment for the entire stream computing
type KisFlow struct {
	// Basic information
	Id   string                // Flow's distributed instance ID (used by KisFlow internally to distinguish different instances)
	Name string                // Flow's readable name
	Conf *config.KisFlowConfig // Flow configuration strategy

	// Function list
	Funcs          map[string]kis.Function // All managed Function objects of the current flow, key: FunctionName
	FlowHead       kis.Function            // Function list head owned by the current Flow
	FlowTail       kis.Function            // Function list tail owned by the current Flow
	flock          sync.RWMutex            // Lock for managing linked list insertion and reading and writing
	ThisFunction   kis.Function            // Currently executing KisFunction object of the Flow
	ThisFunctionId string                  // ID of the currently executing Function
	PrevFunctionId string                  // ID of the Function above the currently executing Function

	// Function list parameters
	funcParams map[string]config.FParam // Custom fixed configuration parameters of the flow in the current Function, Key: KisID of the function instance, value: FParam
	fplock     sync.RWMutex             // Lock for managing funcParams reading and writing

	// Data
	buffer common.KisRowArr  // Internal buffer used to temporarily store input byte data, one data is interface{}, multiple data is []interface{}, i.e., KisBatch
	data   common.KisDataMap // Data sources at various levels of stream computing
	inPut  common.KisRowArr  // Input data for the current Function calculation
}
```
The `Funcs` member here, its key meaning, was previously defined as KisID, and now it should be corrected to mean FunctionName.
Below is a simple correction to the code where `Funcs` member is assigned:

```go
// appendFunc adds Function to Flow, linked list operation
func (flow *KisFlow) appendFunc(function kis.Function, fParam config.FParam) error {
   	// ... ... 


	// Adds the detailed hash relationship of Function Name to the flow object
	flow.Funcs[function.GetConfig().FName] = function

	// ... ... 
}

```

### 6.3.4 Addition in KisPool

> kis-flow/kis/pool.go

```go
// GetFlows gets all the Flows
func (pool *kisPool) GetFlows() []Flow {
	pool.flowLock.RLock() // Read lock
	defer pool.flowLock.RUnlock()

	var flows []Flow

	for _, flow := range pool.flowRouter {
		flows = append(flows, flow)
	}

	return flows
}
```

KisPool adds a method to get all Flows to support the export module.

## 6.4 Configuration Export Unit Testing
Create the `kis_config_export_test.go` file in `kis-flow/test/`:

```go
package test

import (
	"kis-flow/common"
	"kis-flow/file"
	"kis-flow/kis"
	"kis-flow/test/caas"
	"kis-flow/test/faas"
	"testing"
)

func TestConfigExportYmal(t *testing.T) {
	// 0. Register Function callback business
	kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
	kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler)
	kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)

	// 0. Register ConnectorInit and Connector callback business
	kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
	kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)

	// 1. Load configuration file and build Flow
	if err := file.ConfigImportYaml("/Users/gopath/src/kis-flow/test/load_conf/"); err != nil {
		panic(err)
	}

	// 2. Export the configured in-memory KisFlow structure to files
	flows := kis.Pool().GetFlows()
	for _, flow := range flows {
		if err := file.ConfigExportYaml(flow, "/Users/gopath/src/kis-flow/test/export_conf/"); err != nil {
			panic(err)
		}
	}
}

```
Navigate to `kis-flow/test/` and execute:
```bash
go test -test.v -test.paniconexit0 -test.run  TestConfigExportYmal 
```
This will generate exported configurations under `kis-flow/test/export_conf/`:
```bash
├── export_conf
│   ├── conn-ConnName1.yaml
│   ├── flow-flowName1.yaml
│   ├── func-funcName1.yaml
│   ├── func-funcName2.yaml
│   └── func-funcName3.yaml
```

## 6.5 [V0.5] Source Code

https://github.com/aceld/kis-flow/releases/tag/v0.5


---

Author: Aceld
GitHub: https://github.com/aceld

KisFlow Open Source Project Address: https://github.com/aceld/kis-flow

Document: https://github.com/aceld/kis-flow/wiki


---

[Part1-OverView](https://dev.to/aceld/part-1-golang-framework-hands-on-kisflow-streaming-computing-framework-overview-8fh)
[Part2.1-Project Construction / Basic Modules](https://dev.to/aceld/part-2-golang-framework-hands-on-kisflow-streaming-computing-framework-project-construction-basic-modules-cia)
[Part2.2-Project Construction / Basic Modules](https://dev.to/aceld/part-3golang-framework-hands-on-kisflow-stream-computing-framework-project-construction-basic-modules-1epb)
[Part3-Data Stream](https://dev.to/aceld/part-4golang-framework-hands-on-kisflow-stream-computing-framework-data-stream-1mbd)
[Part4-Function Scheduling](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-function-scheduling-4p0h)
[Part5-Connector](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-connector-hcd)
[Part6-Configuration Import and Export](https://dev.to/aceld/part-6golang-framework-hands-on-kisflow-stream-computing-framework-configuration-import-and-export-47o1)
[Part7-KisFlow Action](https://dev.to/aceld/part-7golang-framework-hands-on-kisflow-stream-computing-framework-kisflow-action-3n05)
[Part8-Cache/Params Data Caching and Data Parameters](https://dev.to/aceld/part-8golang-framework-hands-on-cacheparams-data-caching-and-data-parameters-5df5)
[Part9-Multiple Copies of Flow](https://dev.to/aceld/part-8golang-framework-hands-on-multiple-copies-of-flow-c4k)
[Part10-Prometheus Metrics Statistics](https://dev.to/aceld/part-10golang-framework-hands-on-prometheus-metrics-statistics-22f0)
[Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection](https://dev.to/aceld/part-11golang-framework-hands-on-adaptive-registration-of-faas-parameter-types-based-on-reflection-15i9)


---


[Case1-Quick Start](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-quick-start-guide-f51)
[Case2-Flow Parallel Operation](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-flow-parallel-operation-364m)