(Part 5)Golang Framework Hands-on – KisFlow Stream Computing Framework-Connector

package kis

import (
“context”
“kis-flow/config”
)

type Connector interface {
// Init initializes the connection associated with the Connector.
Init() error
// Call invokes the read/write operations of the external storage logic attached to the Connector.
Call(ctx context.Context, flow Flow, args interface{}) error
// GetId gets the ID of the Connector.
GetId() string
// GetName gets the name of the Connector.
GetName() string
}

/*
Connector Init
*/
// ConnInit is the callback function prototype for initializing the third-party mounted storage.
type ConnInit func(conn Connector) error

// connInitRouter is the router managing ConnInit callbacks.
// key: ConnName
type connInitRouter map[string]ConnInit

/*
Connector Call
*/
// CaaS is the callback function prototype for implementing storage read/write operations in the Connector.
type CaaS func(context.Context, Connector, Function, Flow, interface{}) error

// connFuncRouter is the router indexing CaaS callback mappings by FunctionName.
// key: Function Name
// value: CaaS callback for storage read/write operations in the Connector
type connFuncRouter map[string]CaaS

// connSL divides connFuncRouter into two sub-trees based on KisMode
// key: Function KisMode S/L
// value: NsConnRouter
type connSL map[common.KisMode]connFuncRouter

// connTree
// key: Connector Name
// value: connSL secondary tree
type connTree map[string]connSL

// kisPool is used to manage all Function and Flow configurations.
type kisPool struct {
fnRouter funcRouter // All Function management routes
fnLock sync.RWMutex // Lock for fnRouter

flowRouter flowRouter // All flow objects
flowLock sync.RWMutex // Lock for flowRouter

// +++++++++++++++++
cInitRouter connInitRouter // All Connector initialization routes
ciLock sync.RWMutex // Lock for cInitRouter

cTree connTree // All Connector management routes
connectors map[string]Connector // All Connector objects
cLock sync.RWMutex // Lock for cTree
// +++++++++++++++++
}

// Pool is the singleton constructor.
func Pool() *kisPool {
_poolOnce.Do(func() {
// Create a kisPool object
_pool = new(kisPool)

// Initialize fnRouter
_pool.fnRouter = make(funcRouter)

// Initialize flowRouter
_pool.flowRouter = make(flowRouter)

// +++++++++++++++++++++++++
// Initialize cTree
_pool.cTree = make(connTree)
_pool.cInitRouter = make(connInitRouter)
_pool.connectors = make(map[string]Connector)
// +++++++++++++++++++++++++
})

return _pool
}

// Register Connector initialization business.
func (pool *kisPool) RegisterConnectorInit(cname string, c ConnInit) {
pool.ciLock.Lock() // Write lock
defer pool.ciLock.Unlock()

if _, ok := pool.cInitRouter[cname]; !ok {
pool.cInitRouter[cname] = c
} else {
errString := fmt.Sprintf(“KisPool Reg Connector Init Repeat CName=%sn“, cname)
panic(errString)
}

log.Logger().InfoF(“Add KisPool Connector Init CName=%s”, cname)
}

// Dispatch Connector Init
func (pool *kisPool) DispatchConnectorInit(conn Connector) error {
pool.ciLock.RLock() // Read lock
defer pool.ciLock.RUnlock()

init, ok := pool.cInitRouter[conn.GetName()]

if !ok {
panic(errors.New(fmt.Sprintf(“init connector cname = %s not registered..”, conn.GetName())))
}

return init(conn)
}

// Register Connector Call business.
func (pool *kisPool) RegisterConnectorCall(cname string, fname string, mode common.KisMode, c CaaS) {
pool.cLock.Lock() // Write lock
defer pool.cLock.Unlock()

if _, ok := pool.cTree[cname]; !ok {
//cid First registration, does not exist, create a secondary tree NsConnSL
pool.cTree[cname] = make(connSL)

// Initialize various FunctionMode
pool.cTree[cname][common.S] = make(connFuncRouter)
pool.cTree[cname][common.L] = make(connFuncRouter)
}

if _, ok := pool.cTree[cname][mode][fname]; !ok {
pool.cTree[cname][mode][fname] = c
} else {
errString := fmt.Sprintf(“Connector Call Repeat CName=%s, FName=%s, Mode =%sn“, cname, fname, mode)
panic(errString)
}

log.Logger().InfoF(“Add KisPool Connector Call CName=%s, FName=%s, Mode =%s”, cname, fname, mode)
}

// Dispatch Connector Call
func (pool *kisPool) DispatchConnectorCall(ctx context.Context, flow Flow, conn Connector, args interface{}) error {
fn := flow.GetThisFunction()
fnConf := fn.GetConfig()
mode := common.KisMode(fnConf.FMode)

if callback, ok := pool.cTree[conn.GetName()][mode][fnConf.FName]; ok {
return callback(ctx, conn, fn, flow, args)
}

log.Logger().ErrorFX(ctx, “CName:%s FName:%s mode:%s Can not find in KisPool, Not Added.n“, conn.GetName(), fnConf.FName, mode)

return errors.New(fmt.Sprintf(“CName:%s FName:%s mode:%s Can not find in KisPool, Not Added.”, conn.GetName(), fnConf.FName, mode))
}

package conn

import (
“context”
“kis-flow/common”
“kis-flow/config”
“kis-flow/id”
“kis-flow/kis”
“sync”
)

type KisConnector struct {
// Connector ID
CId string
// Connector Name
CName string
// Connector Config
Conf *config.KisConnConfig

// Connector Init
onceInit sync.Once
}

// NewKisConnector creates a KisConnector based on the provided configuration.
func NewKisConnector(config *config.KisConnConfig) *KisConnector {
conn := new(KisConnector)
conn.CId = id.KisID(common.KisIdTypeConnnector)
conn.CName = config.CName
conn.Conf = config

return conn
}

// Init initializes the storage engine connection associated with the Connector.
func (conn *KisConnector) Init() error {
var err error

// Ensure Init is executed only once for a Connector.
conn.onceInit.Do(func() {
err = kis.Pool().DispatchConnectorInit(conn)
})

return err
}

// Call invokes the read/write operations of the external storage logic for the Connector.
func (conn *KisConnector) Call(ctx context.Context, flow kis.Flow, args interface{}) error {
if err := kis.Pool().DispatchConnectorCall(ctx, flow, conn, args); err != nil {
return err
}

return nil
}

func (conn *KisConnector) GetName() string {
return conn.CName
}
func (conn *KisConnector) GetConfig() *config.KisConnConfig {
return conn.Conf
}

func (conn *KisConnector) GetId() string {
return conn.CId
}

package config

import (
“errors”
“fmt”
“kis-flow/common”
)

// KisConnConfig describes the configuration for a KisConnector
type KisConnConfig struct {
// Configuration type
KisType string `yaml:”kistype”`
// Unique identifier
CName string `yaml:”cname”`
// Base storage medium address
AddrString string `yaml:”addrs”`
// Storage medium engine type: “Mysql”, “Redis”, “Kafka”, etc.
Type common.KisConnType `yaml:”type”`
// Identifier for a single storage: e.g., Key name for Redis, Table name for Mysql, Topic name for Kafka, etc.
Key string `yaml:”key”`
// Custom parameters in the configuration information
Params map[string]string `yaml:”params”`
// NsFuncionID bound to storage reading and writing
Load []string `yaml:”load”`
Save []string `yaml:”save”`
}

// NewConnConfig creates a KisConnector strategy configuration object to describe a KisConnector information
func NewConnConfig(cName string, addr string, t common.KisConnType, key string, param FParam) *KisConnConfig {
strategy := new(KisConnConfig)
strategy.CName = cName
strategy.AddrString = addr

strategy.Type = t
strategy.Key = key
strategy.Params = param

return strategy
}

// WithFunc binds Connector with Function
func (cConfig *KisConnConfig) WithFunc(fConfig *KisFuncConfig) error {

switch common.KisMode(fConfig.FMode) {
case common.S:
cConfig.Save = append(cConfig.Save, fConfig.FName)
case common.L:
cConfig.Load = append(cConfig.Load, fConfig.FName)
default:
return errors.New(fmt.Sprintf(“Wrong KisMode %s”, fConfig.FMode))
}

return nil
}

// KisFuncConfig represents a KisFunction strategy configuration
type KisFuncConfig struct {
KisType string `yaml:”kistype”`
FName string `yaml:”fname”`
FMode string `yaml:”fmode”`
Source KisSource `yaml:”source”`
Option KisFuncOption `yaml:”option”`
// ++++++++++
connConf *KisConnConfig
}

func (fConf *KisFuncConfig) AddConnConfig(cConf *KisConnConfig) error {
if cConf == nil {
return errors.New(“KisConnConfig is nil”)
}

// Function needs to be associated with Connector
fConf.connConf = cConf

// Connector needs to be associated with Function
_ = cConf.WithFunc(fConf)

return nil
}

func (fConf *KisFuncConfig) GetConnConfig() (*KisConnConfig, error) {
if fConf.connConf == nil {
return nil, errors.New(“KisFuncConfig.connConf not set”)
}

return fConf.connConf, nil
}

package kis

import (
“context”
“kis-flow/config”
)

// Function represents the basic calculation unit in stream computing. A KisFunction is a basic calculation logic unit in stream computing,
// and any number of KisFunctions can be combined into a KisFlow.
type Function interface {
// Call executes the stream computing logic
Call(ctx context.Context, flow Flow) error

// SetConfig configures the current Function instance
SetConfig(s *config.KisFuncConfig) error
// GetConfig gets the configuration of the current Function instance
GetConfig() *config.KisFuncConfig

// SetFlow sets the Flow instance that the current Function instance depends on
SetFlow(f Flow) error
// GetFlow gets the Flow that the current Function instance depends on
GetFlow() Flow

// ++++++++++++++++++++
// AddConnector adds a Connector to the current Function instance
AddConnector(conn Connector) error
// GetConnector gets the Connector associated with the current Function instance
GetConnector() Connector
// ++++++++++++++++++++

// CreateId generates a random instance KisID for the current Function instance
CreateId()
// GetId gets the FID of the current Function
GetId() string
// GetPrevId gets the FID of the previous Function node on the current Function
GetPrevId() string
// GetNextId gets the FID of the next Function node on the current Function
GetNextId() string

// Next returns the next layer of calculation stream Function, returns nil if the current layer is the last layer
Next() Function
// Prev returns the previous layer of calculation stream Function, returns nil if the current layer is the first layer
Prev() Function
// SetN sets the next Function instance
SetN(f Function)
// SetP sets the previous Function instance
SetP(f Function)
}

type BaseFunction struct {
// Id, the instance ID of KisFunction, used to distinguish different instance objects internally in KisFlow
Id string
Config *config.KisFuncConfig

// flow
flow kis.Flow // Context environment KisFlow

// ++++++++++++++
// connector
connector kis.Connector
// ++++++++++++++

// link
N kis.Function // Next stream computing Function
P kis.Function // Previous stream computing Function
}

// ……..
// ……..

// AddConnector adds a Connector to the current Function instance
func (base *BaseFunction) AddConnector(conn kis.Connector) error {
if conn == nil {
return errors.New(“conn is nil”)
}

base.connector = conn

return nil
}

// GetConnector gets the Connector associated with the current Function instance
func (base *BaseFunction) GetConnector() kis.Connector {
return base.connector
}

package kis

import (
“context”
“kis-flow/common”
“kis-flow/config”
)

type Flow interface {
// Run schedules the Flow, sequentially schedules the Functions in the Flow and executes them
Run(ctx context.Context) error
// Link connects the Functions in the Flow according to the configuration in 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 of the currently executing Function in the Flow
Input() common.KisRowArr
// GetName gets the name of the Flow
GetName() string
// GetThisFunction gets the currently executing Function
GetThisFunction() Function
// GetThisFuncConf gets the configuration of the currently executing Function
GetThisFuncConf() *config.KisFuncConfig
// GetConnector gets the Connector of the currently executing Function
// +++++++++++++++++++++++++++++++++
GetConnector() (Connector, error)
// GetConnConf gets the configuration of the Connector of the currently executing Function
GetConnConf() (*config.KisConnConfig, error)
// +++++++++++++++++++++++++++++++++
}

// GetConnector gets the Connector of the currently executing Function
func (flow *KisFlow) GetConnector() (kis.Connector, error) {
if conn := flow.ThisFunction.GetConnector(); conn != nil {
return conn, nil
} else {
return nil, errors.New(“GetConnector(): Connector is nil”)
}
}

// GetConnConf gets the configuration of the Connector of the currently executing Function
func (flow *KisFlow) GetConnConf() (*config.KisConnConfig, error) {
if conn := flow.ThisFunction.GetConnector(); conn != nil {
return conn.GetConfig(), nil
} else {
return nil, errors.New(“GetConnConf(): Connector is nil”)
}
}

nstype: func
fname: TestFunction_L1
fmode: Load
source:
name: testSource
must:
– stuid
– classid
option:
cname: Test-NsConnector_2

// Link connects the Function to the Flow
// fConf: Current Function strategy
// fParams: Dynamic parameters carried by the current Flow
func (flow *KisFlow) Link(fConf *config.KisFuncConfig, fParams config.FParam) error {
// Create Function instance
f := function.NewKisFunction(flow, fConf)

// ++++++++++++++++++++++++++++++
if fConf.Option.CName != “” {
// The current Function is associated with a Connector and needs to initialize the Connector instance

// Get Connector configuration
connConfig, err := fConf.GetConnConfig()
if err != nil {
panic(err)
}

// Create Connector object
connector := conn.NewKisConnector(connConfig)

// Initialize the Connector, execute the Connector Init method
if err = connector.Init(); err != nil {
panic(err)
}

// Associate the Function instance with the Connector instance
_ = f.AddConnector(connector)
}
// ++++++++++++++++++++++++++++++

// Flow adds Function
if err := flow.appendFunc(f, fParams); err != nil {
return err
}

return nil
}

package test

import (
“context”
“kis-flow/common”
“kis-flow/config”
“kis-flow/flow”
“kis-flow/kis”
“kis-flow/test/caas”
“kis-flow/test/faas”
“testing”
)

func TestNewKisConnector(t *testing.T) {

ctx := context.Background()

// 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. Create 3 KisFunction configuration instances, where myFuncConfig2 has Connector configuration
source1 := config.KisSource{
Name: “Public account Douyin mall customer order data”,
Must: []string{“order_id”, “user_id”},
}

source2 := config.KisSource{
Name: “User order error rate”,
Must: []string{“order_id”, “user_id”},
}

myFuncConfig1 := config.NewFuncConfig(“funcName1”, common.C, &source1, nil)
if myFuncConfig1 == nil {
panic(“myFuncConfig1 is nil”)
}

option := config.KisFuncOption{
CName: “ConnName1”,
}

myFuncConfig2 := config.NewFuncConfig(“funcName2”, common.S, &source2, &option)
if myFuncConfig2 == nil {
panic(“myFuncConfig2 is nil”)
}

myFuncConfig3 := config.NewFuncConfig(“funcName3”, common.E, &source2, nil)
if myFuncConfig3 == nil {
panic(“myFuncConfig3 is nil”)
}

// 2. Create a KisConnector configuration instance
myConnConfig1 := config.NewConnConfig(“ConnName1”, “0.0.0.0:9998”, common.REDIS, “redis-key”, nil)
if myConnConfig1 == nil {
panic(“myConnConfig1 is nil”)
}

// 3. Bind the KisConnector configuration instance to the KisFunction configuration instance
_ = myFuncConfig2.AddConnConfig(myConnConfig1)

// 4. Create a KisFlow configuration instance
myFlowConfig1 := config.NewFlowConfig(“flowName1”, common.FlowEnable)

// 5. Create a KisFlow object
flow1 := flow.NewKisFlow(myFlowConfig1)

// 6. Concatenate Functions to Flow
if err := flow1.Link(myFuncConfig1, nil); err != nil {
panic(err)
}
if err := flow1.Link(myFuncConfig2, nil); err != nil {
panic(err)
}
if err := flow1.Link(myFuncConfig3, nil); err != nil {
panic(err)
}

// 7. Submit original data
_ = flow1.CommitRow(“This is Data1 from Test”)
_ = flow1.CommitRow(“This is Data2 from Test”)
_ = flow1.CommitRow(“This is Data3 from Test”)

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

├── caas
│ ├── caas_demo1.go
│ └── caas_init1.go
├── faas
│ ├── faas_demo1.go
│ ├── faas_demo2.go
│ └── faas_demo3.go

package faas

import (
“context”
“fmt”
“kis-flow/kis”
)

func FuncDemo1Handler(ctx context.Context, flow kis.Flow) error {
fmt.Println(“—> Call funcName1Handler —-“)

for index, row := range flow.Input() {
// Print data
str := fmt.Sprintf(“In FuncName = %s, FuncId = %s, row = %s”, flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
fmt.Println(str)

// Calculate result data
resultStr := fmt.Sprintf(“data from funcName[%s], index = %d”, flow.GetThisFuncConf().FName, index)

// Submit result data
_ = flow.CommitRow(resultStr)
}

return nil
}

package faas

import (
“context”
“fmt”
“kis-flow/kis”
“kis-flow/log”
)

func FuncDemo2Handler(ctx context.Context, flow kis.Flow) error {
fmt.Println(“—> Call funcName2Handler —-“)

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)

conn, err := flow.GetConnector()
if err != nil {
log.Logger().ErrorFX(ctx, “FuncDemo2Handler(): GetConnector err = %sn“, err.Error())
return err
}

if conn.Call(ctx, flow, row) != nil {
log.Logger().ErrorFX(ctx, “FuncDemo2Handler(): Call err = %sn“, err.Error())
return err
}

// Calculate result data
resultStr := fmt.Sprintf(“data from funcName[%s], index = %d”, flow.GetThisFuncConf().FName, index)

// Submit result data
_ = flow.CommitRow(resultStr)
}

return nil
}

package faas

import (
“context”
“fmt”
“kis-flow/kis”
)

func FuncDemo3Handler(ctx context.Context, flow kis.Flow) error {
fmt.Println(“—> Call funcName3Handler —-“)

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

package caas

import (
“fmt”
“kis-flow/kis”
)

func InitConnDemo1(connector kis.Connector) error {
fmt.Println(“===> Call Connector InitDemo1”)
//config info
connConf := connector.GetConfig()

fmt.Println(connConf)

// init connector, such as initializing database connection, etc.

return nil
}

package caas

import (
“context”
“fmt”
“kis-flow/kis”
)

func CaasDemoHanler1(ctx context.Context, conn kis.Connector, fn kis.Function, flow kis.Flow, args interface{}) error {
fmt.Printf(“===> In CaasDemoHanler1: flowName: %s, cName:%s, fnName:%s, mode:%sn“,
flow.GetName(), conn.GetName(), fn.GetConfig().FName, fn.GetConfig().FMode)

fmt.Printf(“===> Call Connector CaasDemoHanler1, args from funciton: %sn“, args)

return nil
}

go test -test.v -test.paniconexit0 -test.run TestNewKisConnector

=== RUN TestNewKisConnector
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
&{ ConnName1 0.0.0.0:9998 redis redis-key map[] [] [funcName2]}
context.Background
====> After CommitSrcData, flow_name = flowName1, flow_id = flow-e5bdad6cb44e47c4b51ffdd4f53148fe
All Level Data =
map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]

KisFunctionC, flow = &{Id:flow-e5bdad6cb44e47c4b51ffdd4f53148fe Name:flowName1 Conf:0xc000026800 Funcs:map[func-66e2b0afa4e14d179aa94c357c412cf8:0xc00007c300 func-f0b4bebf87614828a9375d888c54d13b:0xc00007c2a0 func-f594da0e28da417db6b15ce9c9530f84:0xc00007c240] FlowHead:0xc00007c240 FlowTail:0xc00007c300 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc00007c240 ThisFunctionId:func-f594da0e28da417db6b15ce9c9530f84 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-66e2b0afa4e14d179aa94c357c412cf8:map[] func-f0b4bebf87614828a9375d888c54d13b:map[] func-f594da0e28da417db6b15ce9c9530f84: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-f594da0e28da417db6b15ce9c9530f84, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-f594da0e28da417db6b15ce9c9530f84, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-f594da0e28da417db6b15ce9c9530f84, row = This is Data3 from Test
context.Background
====> After commitCurData, flow_name = flowName1, flow_id = flow-e5bdad6cb44e47c4b51ffdd4f53148fe
All Level Data =
map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-f594da0e28da417db6b15ce9c9530f84:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]

KisFunctionS, flow = &{Id:flow-e5bdad6cb44e47c4b51ffdd4f53148fe Name:flowName1 Conf:0xc000026800 Funcs:map[func-66e2b0afa4e14d179aa94c357c412cf8:0xc00007c300 func-f0b4bebf87614828a9375d888c54d13b:0xc00007c2a0 func-f594da0e28da417db6b15ce9c9530f84:0xc00007c240] FlowHead:0xc00007c240 FlowTail:0xc00007c300 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc00007c2a0 ThisFunctionId:func-f0b4bebf87614828a9375d888c54d13b PrevFunctionId:func-f594da0e28da417db6b15ce9c9530f84 funcParams:map[func-66e2b0afa4e14d179aa94c357c412cf8:map[] func-f0b4bebf87614828a9375d888c54d13b:map[] func-f594da0e28da417db6b15ce9c9530f84: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-f594da0e28da417db6b15ce9c9530f84:[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-f0b4bebf87614828a9375d888c54d13b, 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-f0b4bebf87614828a9375d888c54d13b, 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-f0b4bebf87614828a9375d888c54d13b, 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-e5bdad6cb44e47c4b51ffdd4f53148fe
All Level Data =
map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-f0b4bebf87614828a9375d888c54d13b:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f594da0e28da417db6b15ce9c9530f84:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]

KisFunctionE, flow = &{Id:flow-e5bdad6cb44e47c4b51ffdd4f53148fe Name:flowName1 Conf:0xc000026800 Funcs:map[func-66e2b0afa4e14d179aa94c357c412cf8:0xc00007c300 func-f0b4bebf87614828a9375d888c54d13b:0xc00007c2a0 func-f594da0e28da417db6b15ce9c9530f84:0xc00007c240] FlowHead:0xc00007c240 FlowTail:0xc00007c300 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc00007c300 ThisFunctionId:func-66e2b0afa4e14d179aa94c357c412cf8 PrevFunctionId:func-f0b4bebf87614828a9375d888c54d13b funcParams:map[func-66e2b0afa4e14d179aa94c357c412cf8:map[] func-f0b4bebf87614828a9375d888c54d13b:map[] func-f594da0e28da417db6b15ce9c9530f84: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-f0b4bebf87614828a9375d888c54d13b:[data from funcName[funcName2], index = 0 data from funcName[funcName2], index = 1 data from funcName[funcName2], index = 2] func-f594da0e28da417db6b15ce9c9530f84:[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]}

—> Call funcName3Handler —-
In FuncName = funcName3, FuncId = func-66e2b0afa4e14d179aa94c357c412cf8, row = data from funcName[funcName2], index = 0
In FuncName = funcName3, FuncId = func-66e2b0afa4e14d179aa94c357c412cf8, row = data from funcName[funcName2], index = 1
In FuncName = funcName3, FuncId = func-66e2b0afa4e14d179aa94c357c412cf8, row = data from funcName[funcName2], index = 2
— PASS: TestNewKisConnector (0.00s)
PASS