添加pipeline

go.mod

@@ -2,7 +2,12 @@ module git.sxidc.com/go-tools/utils
 
 go 1.21.3
 
-require github.com/redis/go-redis/v9 v9.4.0
+require (
+	github.com/fatih/structs v1.1.0
+	github.com/mitchellh/mapstructure v1.5.0
+	github.com/redis/go-redis/v9 v9.4.0
+	gopkg.in/yaml.v3 v3.0.1
+)
 
 require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect

go.sum

@@ -1,3 +1,5 @@
+git.sxidc.com/go-tools/pipeline v0.0.0-20231107123717-990956b855f8 h1:FjhX1I8cjoaaW5jSaZbnQ7zilCLjwksVUoVVXkNnOr0=
+git.sxidc.com/go-tools/pipeline v0.0.0-20231107123717-990956b855f8/go.mod h1:0edkUfJ52WCL8dXlsn5gS968gx1469Ge2ccBmQy2yag=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -6,5 +8,13 @@ github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/fatih/structs v1.1.0 h1:Q7juDM0QtcnhCpeyLGQKyg4TOIghuNXrkL32pHAUMxo=
+github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
+github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
+github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/redis/go-redis/v9 v9.4.0 h1:Yzoz33UZw9I/mFhx4MNrB6Fk+XHO1VukNcCa1+lwyKk=
 github.com/redis/go-redis/v9 v9.4.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pipeline/component/base.go

@@ -0,0 +1,61 @@
+package component
+
+import (
+	"fmt"
+)
+
+type BaseComponent struct {
+	typeName  string
+	name      string
+	runParams map[string]any
+}
+
+func NewBaseComponent(typeName string, name string, runParams map[string]any) *BaseComponent {
+	return &BaseComponent{
+		typeName:  typeName,
+		name:      name,
+		runParams: runParams,
+	}
+}
+
+func (c *BaseComponent) GetType() string {
+	return c.typeName
+}
+
+func (c *BaseComponent) GetName() string {
+	return c.name
+}
+
+type RunFunc func(globalRunParams *GlobalRunParams, dynamicParams map[string]any) (any, error)
+
+func (c *BaseComponent) OnRun(globalRunParams *GlobalRunParams, dynamicParams map[string]any, runFunc RunFunc) (any, error) {
+	if globalRunParams == nil {
+		globalRunParams = new(GlobalRunParams)
+	}
+
+	fmt.Println("----------" + c.typeName + "::" + c.name + "开始运行----------")
+
+	if dynamicParams == nil {
+		dynamicParams = make(map[string]any)
+	}
+
+	for key, params := range c.runParams {
+		_, ok := dynamicParams[key]
+		if !ok {
+			dynamicParams[key] = params
+		}
+	}
+
+	result, err := runFunc(globalRunParams, dynamicParams)
+	if err != nil {
+		return nil, err
+	}
+
+	if result != nil {
+		globalRunParams.SetLastResult(result)
+	}
+
+	fmt.Println("----------" + c.typeName + "类型的" + c.name + "结束运行----------")
+
+	return result, nil
+}

pipeline/component/component.go

@@ -0,0 +1,83 @@
+package component
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/utils"
+	"sync"
+)
+
+type Component interface {
+	GetType() string
+	GetName() string
+	Run(globalRunParams *GlobalRunParams, dynamicParams map[string]any) (any, error)
+}
+
+type Builder interface {
+	ProductType() string
+	Build(name string, buildParams map[string]any, runParams map[string]any) (Component, error)
+}
+
+var nodeBuilderMap = sync.Map{}
+
+func RegisterComponentBuilders(builders ...Builder) error {
+	if builders == nil || len(builders) == 0 {
+		return nil
+	}
+
+	for _, builder := range builders {
+		nodeType := builder.ProductType()
+		if utils.IsStringEmpty(nodeType) {
+			return errors.New("节点类型名称不能为空")
+		}
+
+		_, ok := nodeBuilderMap.Load(nodeType)
+		if ok {
+			return errors.New(nodeType + "的builder已注册")
+		}
+
+		nodeBuilderMap.Store(nodeType, builder)
+	}
+
+	return nil
+}
+
+func UnRegisterComponents(typeNames []string) {
+	if typeNames == nil || len(typeNames) == 0 {
+		return
+	}
+
+	for _, typeName := range typeNames {
+		nodeBuilderMap.Delete(typeName)
+	}
+}
+
+func BuildComponent(typeName string, name string, buildParams map[string]any, runParams map[string]any) (Component, error) {
+	if utils.IsStringEmpty(typeName) {
+		return nil, errors.New("没有传递类型名称")
+	}
+
+	if utils.IsStringEmpty(name) {
+		return nil, errors.New("没有传递名称")
+	}
+
+	value, ok := nodeBuilderMap.Load(typeName)
+	if !ok {
+		return nil, errors.New(typeName + "未注册")
+	}
+
+	builder, ok := value.(Builder)
+	if !ok {
+		return nil, errors.New(typeName + "类型转换失败")
+	}
+
+	newComponent, err := builder.Build(name, buildParams, runParams)
+	if err != nil {
+		return nil, err
+	}
+
+	if newComponent == nil {
+		return nil, errors.New(typeName + "的builder返回的对象为空")
+	}
+
+	return newComponent, nil
+}

pipeline/component/flow/bi.go

@@ -0,0 +1,166 @@
+package flow
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypeBi = "bi"
+)
+
+const (
+	BiLeft  = "bi_left"
+	BiRight = "bi_right"
+)
+
+type BiBuildParams struct {
+	Components []SubComponentBuildParams `mapstructure:"components" structs:"components"`
+}
+
+func (params *BiBuildParams) Check() error {
+	if params.Components == nil || len(params.Components) == 0 {
+		return errors.New(TypeBi + "流没有包含任何组件")
+	}
+
+	subComponentMap := make(map[string]bool)
+	for _, componentParams := range params.Components {
+		err := componentParams.Check(TypeBi)
+		if err != nil {
+			return err
+		}
+
+		_, ok := subComponentMap[componentParams.Name]
+		if ok {
+			return errors.New("该流程中存在同名的组件: " + componentParams.Name)
+		}
+
+		subComponentMap[componentParams.Name] = true
+	}
+
+	return nil
+}
+
+type BiRunParams struct {
+	IsBi        bool `mapstructure:"is_bi" structs:"is_bi"`
+	LeftParams  any  `mapstructure:"left_params" structs:"left_params"`
+	RightParams any  `mapstructure:"right_params" structs:"right_params"`
+}
+
+func (params *BiRunParams) Check() error {
+	if params.LeftParams == nil {
+		return errors.New(TypeBi + "没有传递左参数")
+	}
+
+	if params.LeftParams == nil {
+		return errors.New(TypeBi + "没有传递右参数")
+	}
+
+	return nil
+}
+
+type Bi struct {
+	component.BaseComponent
+	Components []component.Component
+}
+
+func (f *Bi) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return f.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			params := new(BiRunParams)
+			err := mapstructure.Decode(dynamicParams, params)
+			if err != nil {
+				return nil, errors.New(TypeBi + "流程运行时参数Decode失败: " + err.Error())
+			}
+
+			err = params.Check()
+			if err != nil {
+				return nil, err
+			}
+
+			result, err := f.runSubComponents(globalRunParams, dynamicParams, params.LeftParams, params.RightParams)
+			if err != nil {
+				return nil, err
+			}
+
+			if !params.IsBi {
+				return result, nil
+			}
+
+			result, err = f.runSubComponents(globalRunParams, dynamicParams, params.RightParams, params.LeftParams)
+			if err != nil {
+				return nil, err
+			}
+
+			return result, nil
+		})
+}
+
+func (f *Bi) runSubComponents(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any,
+	leftParams any, rightParams any) (any, error) {
+	var result any
+
+	for _, subComponent := range f.Components {
+		subDynamicParamsMap := make(map[string]any)
+		subDynamicParams, ok := dynamicParams[subComponent.GetName()]
+		if ok {
+			innerSubDynamicParamsMap, ok := subDynamicParams.(map[string]any)
+			if !ok {
+				return nil, errors.New(TypeBi + "流程" + f.GetName() + "的子组件" + subComponent.GetName() + "动态参数类型错误")
+			}
+
+			subDynamicParamsMap = innerSubDynamicParamsMap
+		}
+
+		subDynamicParamsMap[BiLeft] = leftParams
+		subDynamicParamsMap[BiRight] = rightParams
+
+		innerResult, err := subComponent.Run(globalRunParams, subDynamicParamsMap)
+		if err != nil {
+			return nil, err
+		}
+
+		result = innerResult
+	}
+
+	return result, nil
+}
+
+type BiBuilder struct{}
+
+func (builder *BiBuilder) ProductType() string {
+	return TypeBi
+}
+
+func (builder *BiBuilder) Build(name string, buildParams map[string]any, runParams map[string]any) (component.Component, error) {
+	if buildParams == nil || len(buildParams) == 0 {
+		return nil, errors.New(TypeBi + "流程没有传递构建参数")
+	}
+
+	flowBuildParams := new(BiBuildParams)
+	err := mapstructure.Decode(buildParams, flowBuildParams)
+	if err != nil {
+		return nil, errors.New(TypeBi + "流程构建参数Decode失败: " + err.Error())
+	}
+
+	err = flowBuildParams.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	subComponents := make([]component.Component, 0)
+	for _, componentParams := range flowBuildParams.Components {
+		subComponent, err := componentParams.BuildComponent()
+		if err != nil {
+			return nil, err
+		}
+
+		subComponents = append(subComponents, subComponent)
+	}
+
+	return &Bi{
+		BaseComponent: *component.NewBaseComponent(TypeBi, name, runParams),
+		Components:    subComponents,
+	}, nil
+}

pipeline/component/flow/flow.go

@@ -0,0 +1,30 @@
+package flow
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/utils"
+)
+
+type SubComponentBuildParams struct {
+	Type        string         `mapstructure:"type" structs:"type"`
+	Name        string         `mapstructure:"name" structs:"name"`
+	BuildParams map[string]any `mapstructure:"build_params" structs:"build_params"`
+	RunParams   map[string]any `mapstructure:"run_params" structs:"run_params"`
+}
+
+func (params *SubComponentBuildParams) Check(parentNodeType string) error {
+	if utils.IsStringEmpty(params.Type) {
+		return errors.New(parentNodeType + "没有传递子节点类型")
+	}
+
+	if utils.IsStringEmpty(params.Name) {
+		return errors.New(parentNodeType + "没有传递子节点名称")
+	}
+
+	return nil
+}
+
+func (params *SubComponentBuildParams) BuildComponent() (component.Component, error) {
+	return component.BuildComponent(params.Type, params.Name, params.BuildParams, params.RunParams)
+}

pipeline/component/flow/if.go

@@ -0,0 +1,178 @@
+package flow
+
+import (
+	"errors"
+	"fmt"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypeIf = "if"
+)
+
+type IfBuildParams struct {
+	ConditionComponent *SubComponentBuildParams `mapstructure:"condition" structs:"condition"`
+	TrueComponent      *SubComponentBuildParams `mapstructure:"condition_true" structs:"condition_true"`
+	FalseComponent     *SubComponentBuildParams `mapstructure:"condition_false" structs:"condition_false"`
+}
+
+func (params *IfBuildParams) Check() error {
+	if params.ConditionComponent == nil {
+		return errors.New(TypeIf + "流程的条件参数为空")
+	}
+
+	if params.TrueComponent == nil {
+		return errors.New(TypeIf + "流程的真组件参数为空")
+	}
+
+	if params.FalseComponent == nil {
+		return errors.New(TypeIf + "流程的假组件参数为空")
+	}
+
+	err := params.ConditionComponent.Check(TypeIf)
+	if err != nil {
+		return err
+	}
+
+	err = params.TrueComponent.Check(TypeIf)
+	if err != nil {
+		return err
+	}
+
+	if params.ConditionComponent.Name == params.TrueComponent.Name ||
+		params.TrueComponent.Name == params.FalseComponent.Name ||
+		params.ConditionComponent.Name == params.FalseComponent.Name {
+		return fmt.Errorf("流程中存在同名的组件: condition: %s true: %s false: %s",
+			params.ConditionComponent.Name, params.TrueComponent.Name, params.FalseComponent.Name)
+	}
+
+	return nil
+}
+
+type If struct {
+	component.BaseComponent
+	ConditionComponent component.Component
+	TrueComponent      component.Component
+	FalseComponent     component.Component
+}
+
+func (f *If) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return f.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			lastResult := globalRunParams.GetLastResult()
+
+			var conditionDynamicParamsMap map[string]any
+			conditionDynamicParams, ok := dynamicParams[f.ConditionComponent.GetName()]
+			if ok {
+				innerConditionDynamicParamsMap, ok := conditionDynamicParams.(map[string]any)
+				if !ok {
+					return nil, errors.New(TypeIf + "流程" + f.GetName() + "的条件组件" + f.ConditionComponent.GetName() + "动态参数类型错误")
+				}
+
+				conditionDynamicParamsMap = innerConditionDynamicParamsMap
+			}
+
+			conditionResult, err := f.ConditionComponent.Run(globalRunParams, conditionDynamicParamsMap)
+			if err != nil {
+				return nil, err
+			}
+
+			condition, ok := conditionResult.(bool)
+			if !ok {
+				return nil, errors.New(TypeIf + "流程的条件子组件返回结果应当为bool")
+			}
+
+			if condition {
+				var trueDynamicParamsMap map[string]any
+				trueDynamicParams, ok := dynamicParams[f.TrueComponent.GetName()]
+				if ok {
+					innerTrueDynamicParamsMap, ok := trueDynamicParams.(map[string]any)
+					if !ok {
+						return nil, errors.New(TypeIf + "流程" + f.GetName() + "的真组件" + f.TrueComponent.GetName() + "动态参数类型错误")
+					}
+
+					trueDynamicParamsMap = innerTrueDynamicParamsMap
+				}
+
+				result, err := f.TrueComponent.Run(globalRunParams, trueDynamicParamsMap)
+				if err != nil {
+					return nil, err
+				}
+
+				return result, nil
+			} else {
+				if f.FalseComponent != nil {
+					var falseDynamicParamsMap map[string]any
+					falseDynamicParams, ok := dynamicParams[f.FalseComponent.GetName()]
+					if ok {
+						innerFalseDynamicParamsMap, ok := falseDynamicParams.(map[string]any)
+						if !ok {
+							return nil, errors.New(TypeIf + "流程" + f.GetName() + "的假组件" + f.FalseComponent.GetName() + "动态参数类型错误")
+						}
+
+						falseDynamicParamsMap = innerFalseDynamicParamsMap
+					}
+
+					result, err := f.FalseComponent.Run(globalRunParams, falseDynamicParamsMap)
+					if err != nil {
+						return nil, err
+					}
+
+					return result, nil
+				}
+			}
+
+			return lastResult, nil
+		})
+}
+
+type IfBuilder struct{}
+
+func (builder *IfBuilder) ProductType() string {
+	return TypeIf
+}
+
+func (builder *IfBuilder) Build(name string, buildParams map[string]any, runParams map[string]any) (component.Component, error) {
+	if buildParams == nil || len(buildParams) == 0 {
+		return nil, errors.New(TypeIf + "流没有传递构建参数")
+	}
+
+	flowBuildParams := new(IfBuildParams)
+	err := mapstructure.Decode(buildParams, flowBuildParams)
+	if err != nil {
+		return nil, errors.New(TypeIf + "流程构建参数Decode失败: " + err.Error())
+	}
+
+	err = flowBuildParams.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	conditionFlow, err := flowBuildParams.ConditionComponent.BuildComponent()
+	if err != nil {
+		return nil, err
+	}
+
+	trueFlow, err := flowBuildParams.TrueComponent.BuildComponent()
+	if err != nil {
+		return nil, err
+	}
+
+	var falseFlow component.Component
+	if flowBuildParams.FalseComponent != nil {
+		innerFalseFlow, err := flowBuildParams.FalseComponent.BuildComponent()
+		if err != nil {
+			return nil, err
+		}
+
+		falseFlow = innerFalseFlow
+	}
+
+	return &If{
+		BaseComponent:      *component.NewBaseComponent(TypeIf, name, runParams),
+		ConditionComponent: conditionFlow,
+		TrueComponent:      trueFlow,
+		FalseComponent:     falseFlow,
+	}, nil
+}

pipeline/component/flow/loop.go

@@ -0,0 +1,159 @@
+package flow
+
+import (
+	"errors"
+	"fmt"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypeLoop = "loop"
+)
+
+type LoopBuildParams struct {
+	ConditionComponent *SubComponentBuildParams `mapstructure:"condition" structs:"condition"`
+	SubComponent       *SubComponentBuildParams `mapstructure:"sub" structs:"sub"`
+}
+
+func (params *LoopBuildParams) Check() error {
+	if params.ConditionComponent == nil {
+		return errors.New(TypeLoop + "流程的条件组件参数为空")
+	}
+
+	if params.SubComponent == nil {
+		return errors.New(TypeLoop + "流程的子组件参数为空")
+	}
+
+	err := params.ConditionComponent.Check(TypeLoop)
+	if err != nil {
+		return err
+	}
+
+	err = params.SubComponent.Check(TypeLoop)
+	if err != nil {
+		return err
+	}
+
+	if params.ConditionComponent.Name == params.SubComponent.Name {
+		return fmt.Errorf("流程中存在同名的组件: condition: %s flow: %s",
+			params.ConditionComponent.Name, params.SubComponent.Name)
+	}
+
+	return nil
+}
+
+type Loop struct {
+	component.BaseComponent
+	ConditionComponent component.Component
+	SubComponent       component.Component
+}
+
+func (f *Loop) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return f.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			lastResult := globalRunParams.GetLastResult()
+
+			var conditionDynamicParamsMap map[string]any
+			conditionDynamicParams, ok := dynamicParams[f.ConditionComponent.GetName()]
+			if ok {
+				innerConditionDynamicParamsMap, ok := conditionDynamicParams.(map[string]any)
+				if !ok {
+					return nil, errors.New(TypeLoop + "流程" + f.GetName() + "的条件组件" + f.ConditionComponent.GetName() + "动态参数类型错误")
+				}
+
+				conditionDynamicParamsMap = innerConditionDynamicParamsMap
+			}
+
+			var subDynamicParamsMap map[string]any
+			subDynamicParams, ok := dynamicParams[f.SubComponent.GetName()]
+			if ok {
+				innerSubDynamicParamsMap, ok := subDynamicParams.(map[string]any)
+				if !ok {
+					return nil, errors.New(TypeLoop + "流程" + f.GetName() + "的子组件" + f.SubComponent.GetName() + "动态参数类型错误")
+				}
+
+				subDynamicParamsMap = innerSubDynamicParamsMap
+			}
+
+			condition, err := f.runConditionComponent(globalRunParams, conditionDynamicParamsMap)
+			if err != nil {
+				return nil, err
+			}
+
+			if !condition {
+				return lastResult, nil
+			}
+
+			var result any
+			for condition {
+				innerResult, err := f.SubComponent.Run(globalRunParams, subDynamicParamsMap)
+				if err != nil {
+					return nil, err
+				}
+
+				innerCondition, err := f.runConditionComponent(globalRunParams, conditionDynamicParamsMap)
+				if err != nil {
+					return nil, err
+				}
+
+				condition = innerCondition
+				result = innerResult
+			}
+
+			return result, nil
+		})
+}
+
+func (f *Loop) runConditionComponent(globalRunParams *component.GlobalRunParams, conditionDynamicParamsMap map[string]any) (bool, error) {
+	conditionResult, err := f.ConditionComponent.Run(globalRunParams, conditionDynamicParamsMap)
+	if err != nil {
+		return false, err
+	}
+
+	condition, ok := conditionResult.(bool)
+	if !ok {
+		return false, errors.New(TypeLoop + "流程的条件子组件返回结果应当为bool")
+	}
+
+	return condition, nil
+}
+
+type LoopBuilder struct{}
+
+func (builder *LoopBuilder) ProductType() string {
+	return TypeLoop
+}
+
+func (builder *LoopBuilder) Build(name string, buildParams map[string]any, runParams map[string]any) (component.Component, error) {
+	if buildParams == nil || len(buildParams) == 0 {
+		return nil, errors.New(TypeLoop + "流程没有传递构建参数")
+	}
+
+	flowBuildParams := new(LoopBuildParams)
+	err := mapstructure.Decode(buildParams, flowBuildParams)
+	if err != nil {
+		return nil, errors.New(TypeLoop + "流程构建参数Decode失败: " + err.Error())
+	}
+
+	err = flowBuildParams.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	conditionFlow, err := flowBuildParams.ConditionComponent.BuildComponent()
+	if err != nil {
+		return nil, err
+	}
+
+	subFlow, err := flowBuildParams.SubComponent.BuildComponent()
+	if err != nil {
+		return nil, err
+	}
+
+	return &Loop{
+		BaseComponent:      *component.NewBaseComponent(TypeLoop, name, runParams),
+		ConditionComponent: conditionFlow,
+		SubComponent:       subFlow,
+	}, nil
+}

pipeline/component/flow/range.go

@@ -0,0 +1,111 @@
+package flow
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypeRange = "range"
+)
+
+const (
+	RangeIndex     = "range_index"
+	RangeOnceValue = "range_once_value"
+)
+
+type RangeBuildParams struct {
+	Values       []any                    `mapstructure:"values" structs:"values"`
+	SubComponent *SubComponentBuildParams `mapstructure:"sub" structs:"sub"`
+}
+
+func (params *RangeBuildParams) Check() error {
+	if params.SubComponent == nil {
+		return errors.New(TypeRange + "流程的子组件参数为空")
+	}
+
+	err := params.SubComponent.Check(TypeRange)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+type Range struct {
+	component.BaseComponent
+	Values       []any
+	SubComponent component.Component
+}
+
+func (f *Range) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return f.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			lastResult := globalRunParams.GetLastResult()
+
+			subDynamicParamsMap := make(map[string]any)
+			subDynamicParams, ok := dynamicParams[f.SubComponent.GetName()]
+			if ok {
+				innerSubDynamicParamsMap, ok := subDynamicParams.(map[string]any)
+				if !ok {
+					return nil, errors.New(TypeLoop + "流程" + f.GetName() + "的子组件" + f.SubComponent.GetName() + "动态参数类型错误")
+				}
+
+				subDynamicParamsMap = innerSubDynamicParamsMap
+			}
+
+			if f.Values == nil || len(f.Values) == 0 {
+				return lastResult, nil
+			}
+
+			var result any
+			for index, value := range f.Values {
+				subDynamicParamsMap[RangeIndex] = index
+				subDynamicParamsMap[RangeOnceValue] = value
+
+				innerResult, err := f.SubComponent.Run(globalRunParams, subDynamicParamsMap)
+				if err != nil {
+					return nil, err
+				}
+
+				result = innerResult
+			}
+
+			return result, nil
+		})
+}
+
+type RangeBuilder struct{}
+
+func (builder *RangeBuilder) ProductType() string {
+	return TypeRange
+}
+
+func (builder *RangeBuilder) Build(name string, buildParams map[string]any, runParams map[string]any) (component.Component, error) {
+	if buildParams == nil || len(buildParams) == 0 {
+		return nil, errors.New(TypeRange + "流程没有传递构建参数")
+	}
+
+	flowBuildParams := new(RangeBuildParams)
+	err := mapstructure.Decode(buildParams, flowBuildParams)
+	if err != nil {
+		return nil, errors.New(TypeRange + "流程构建参数Decode失败: " + err.Error())
+	}
+
+	err = flowBuildParams.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	subFlow, err := flowBuildParams.SubComponent.BuildComponent()
+	if err != nil {
+		return nil, err
+	}
+
+	return &Range{
+		BaseComponent: *component.NewBaseComponent(TypeLoop, name, runParams),
+		Values:        flowBuildParams.Values,
+		SubComponent:  subFlow,
+	}, nil
+}

pipeline/component/flow/seq.go

@@ -0,0 +1,109 @@
+package flow
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypeSeq = "seq"
+)
+
+type SeqBuildParams struct {
+	Components []SubComponentBuildParams `mapstructure:"components" structs:"components"`
+}
+
+func (params *SeqBuildParams) Check() error {
+	if params.Components == nil || len(params.Components) == 0 {
+		return errors.New(TypeSeq + "流没有包含任何组件")
+	}
+
+	subComponentMap := make(map[string]bool)
+	for _, componentParams := range params.Components {
+		err := componentParams.Check(TypeSeq)
+		if err != nil {
+			return err
+		}
+
+		_, ok := subComponentMap[componentParams.Name]
+		if ok {
+			return errors.New("该流程中存在同名的组件: " + componentParams.Name)
+		}
+
+		subComponentMap[componentParams.Name] = true
+	}
+
+	return nil
+}
+
+type Seq struct {
+	component.BaseComponent
+	Components []component.Component
+}
+
+func (f *Seq) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return f.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			var result any
+			for _, subComponent := range f.Components {
+				var subDynamicParamsMap map[string]any
+				subDynamicParams, ok := dynamicParams[subComponent.GetName()]
+				if ok {
+					innerSubDynamicParamsMap, ok := subDynamicParams.(map[string]any)
+					if !ok {
+						return nil, errors.New(TypeSeq + "流程" + f.GetName() + "的子组件" + subComponent.GetName() + "动态参数类型错误")
+					}
+
+					subDynamicParamsMap = innerSubDynamicParamsMap
+				}
+
+				innerResult, err := subComponent.Run(globalRunParams, subDynamicParamsMap)
+				if err != nil {
+					return nil, err
+				}
+
+				result = innerResult
+			}
+
+			return result, nil
+		})
+}
+
+type SeqBuilder struct{}
+
+func (builder *SeqBuilder) ProductType() string {
+	return TypeSeq
+}
+
+func (builder *SeqBuilder) Build(name string, buildParams map[string]any, runParams map[string]any) (component.Component, error) {
+	if buildParams == nil || len(buildParams) == 0 {
+		return nil, errors.New(TypeSeq + "流程没有传递构建参数")
+	}
+
+	flowBuildParams := new(SeqBuildParams)
+	err := mapstructure.Decode(buildParams, flowBuildParams)
+	if err != nil {
+		return nil, errors.New(TypeSeq + "流程构建参数Decode失败: " + err.Error())
+	}
+
+	err = flowBuildParams.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	subComponents := make([]component.Component, 0)
+	for _, componentParams := range flowBuildParams.Components {
+		subComponent, err := componentParams.BuildComponent()
+		if err != nil {
+			return nil, err
+		}
+
+		subComponents = append(subComponents, subComponent)
+	}
+
+	return &Seq{
+		BaseComponent: *component.NewBaseComponent(TypeSeq, name, runParams),
+		Components:    subComponents,
+	}, nil
+}

pipeline/component/global_run_params.go

@@ -0,0 +1,14 @@
+package component
+
+type GlobalRunParams struct {
+	CustomRunParams map[string]any
+	lastResult      any
+}
+
+func (params GlobalRunParams) SetLastResult(value any) {
+	params.lastResult = value
+}
+
+func (params GlobalRunParams) GetLastResult() any {
+	return params.lastResult
+}

pipeline/definition.go

@@ -0,0 +1,101 @@
+package pipeline
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/component/flow"
+	"git.sxidc.com/go-tools/utils/pipeline/utils"
+	"github.com/fatih/structs"
+)
+
+type Definition struct {
+	Name       string                `yaml:"name"`
+	Components []ComponentDefinition `yaml:"components"`
+}
+
+func (def *Definition) Check() error {
+	if utils.IsStringEmpty(def.Name) {
+		return errors.New("流水线没有传递名称")
+	}
+
+	if def.Components == nil || len(def.Components) == 0 {
+		return errors.New("流水线没有添加组件")
+	}
+
+	for _, c := range def.Components {
+		err := c.Check()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+type ComponentDefinition struct {
+	Type        string         `yaml:"type"`
+	Name        string         `yaml:"name"`
+	BuildParams map[string]any `yaml:"build_params"`
+	RunParams   map[string]any `yaml:"run_params"`
+}
+
+func (def *ComponentDefinition) Check() error {
+	if utils.IsStringEmpty(def.Type) {
+		return errors.New("组件没有传递类型")
+	}
+
+	if utils.IsStringEmpty(def.Name) {
+		return errors.New("组件没有传递名称")
+	}
+
+	return nil
+}
+
+func (def *Definition) NewPipeline() (*Pipeline, error) {
+	err := def.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	subComponentsParams := make([]flow.SubComponentBuildParams, 0)
+	for _, subDef := range def.Components {
+		subComponentsParams = append(subComponentsParams, flow.SubComponentBuildParams{
+			Type:        subDef.Type,
+			Name:        subDef.Name,
+			BuildParams: subDef.BuildParams,
+			RunParams:   subDef.RunParams,
+		})
+	}
+
+	pipeline, err := newPipelineWithBuildParams(def.Name, &flow.SeqBuildParams{
+		Components: subComponentsParams,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return pipeline, nil
+}
+
+func newPipelineWithBuildParams(name string, params *flow.SeqBuildParams) (*Pipeline, error) {
+	if utils.IsStringEmpty(name) {
+		return nil, errors.New("没有传递流水线的名称")
+	}
+
+	err := params.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	rootComponent, err := component.BuildComponent("seq", name, structs.Map(params), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	rootSeqFlow, ok := rootComponent.(*flow.Seq)
+	if !ok {
+		return nil, errors.New("流水线的根流程不是顺序流程")
+	}
+
+	return &Pipeline{Flow: rootSeqFlow}, nil
+}

pipeline/pipeline.go

@@ -0,0 +1,116 @@
+package pipeline
+
+import (
+	"errors"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/component/flow"
+	"git.sxidc.com/go-tools/utils/pipeline/utils"
+	"gopkg.in/yaml.v3"
+	"os"
+)
+
+func init() {
+	err := component.RegisterComponentBuilders(
+		&flow.SeqBuilder{}, &flow.IfBuilder{}, &flow.LoopBuilder{}, &flow.RangeBuilder{}, &flow.BiBuilder{},
+	)
+	if err != nil {
+		panic(err)
+	}
+}
+
+type Pipeline struct {
+	Flow *flow.Seq
+}
+
+func (p *Pipeline) Run(globalRunParams map[string]any, dynamicParams map[string]any) *RunToken {
+	token := NewRunToken()
+
+	go func() {
+		result, err := p.Flow.Run(&component.GlobalRunParams{
+			CustomRunParams: globalRunParams,
+		}, dynamicParams)
+		if err != nil {
+			token.Err = err
+			return
+		}
+
+		token.Result = result
+		token.Done()
+	}()
+
+	return token
+}
+
+func NewPipelineFromYaml(yamlPath string) (*Pipeline, error) {
+	if utils.IsStringEmpty(yamlPath) {
+		return nil, errors.New("没有传递流水线定义文件")
+	}
+
+	if !utils.PathExists(yamlPath) {
+		return nil, errors.New("流水线定义文件不存在")
+	}
+
+	yamlBytes, err := os.ReadFile(yamlPath)
+	if err != nil {
+		return nil, errors.New("读取流水线定义文件失败: " + err.Error())
+	}
+
+	def := new(Definition)
+	err = yaml.Unmarshal(yamlBytes, def)
+	if err != nil {
+		return nil, errors.New("Unmarshal流水线定义文件失败: " + err.Error())
+	}
+
+	return def.NewPipeline()
+}
+
+func NewPipelineFromYamlStr(yamlStr string) (*Pipeline, error) {
+	if utils.IsStringEmpty(yamlStr) {
+		return nil, errors.New("没有传递流水线定义")
+	}
+
+	def := new(Definition)
+	err := yaml.Unmarshal([]byte(yamlStr), def)
+	if err != nil {
+		return nil, errors.New("Unmarshal流水线定义文件失败: " + err.Error())
+	}
+
+	return def.NewPipeline()
+}
+
+func LoadDynamicParamsFromYaml(yamlPath string) (map[string]any, error) {
+	if utils.IsStringEmpty(yamlPath) {
+		return nil, errors.New("没有传递流水线动态参数文件不存在")
+	}
+
+	if !utils.PathExists(yamlPath) {
+		return nil, errors.New("流水线动态参数文件不存在")
+	}
+
+	yamlBytes, err := os.ReadFile(yamlPath)
+	if err != nil {
+		return nil, errors.New("读取流水线动态参数文件失败: " + err.Error())
+	}
+
+	dynamicParams := make(map[string]any)
+	err = yaml.Unmarshal(yamlBytes, dynamicParams)
+	if err != nil {
+		return nil, err
+	}
+
+	return dynamicParams, nil
+}
+
+func LoadDynamicParamsFromYamlStr(yamlStr string) (map[string]any, error) {
+	if utils.IsStringEmpty(yamlStr) {
+		return nil, errors.New("没有传递流水线动态参数不存在")
+	}
+
+	dynamicParams := make(map[string]any)
+	err := yaml.Unmarshal([]byte(yamlStr), dynamicParams)
+	if err != nil {
+		return nil, err
+	}
+
+	return dynamicParams, nil
+}

pipeline/run_token.go

@@ -0,0 +1,37 @@
+package pipeline
+
+type RunToken struct {
+	Result   any
+	Err      error
+	doneChan chan any
+}
+
+func NewRunToken() *RunToken {
+	return &RunToken{
+		Result:   nil,
+		Err:      nil,
+		doneChan: make(chan any),
+	}
+}
+
+func (token *RunToken) Wait() {
+	if token.doneChan == nil {
+		return
+	}
+
+	for {
+		select {
+		case <-token.doneChan:
+			return
+		}
+	}
+}
+
+func (token *RunToken) Done() {
+	if token.doneChan == nil {
+		return
+	}
+
+	close(token.doneChan)
+	token.doneChan = nil
+}

pipeline/test/def.yaml

@@ -0,0 +1,81 @@
+name: test
+components:
+  - type: seq
+    name: seq-flow
+    build_params:
+      components:
+        - type: println
+          name: seq-flow-node
+          run_params:
+            content: seq-flow-node
+  - type: println
+    name: pipeline-node
+    run_params:
+      content: pipeline-node
+  - type: if
+    name: if-flow
+    build_params:
+      condition:
+        type: seq
+        name: condition_flow
+        build_params:
+          components:
+            - type: bool
+              name: bool
+              run_params:
+                op: pass
+                value: true
+      condition_true:
+        type: seq
+        name: true-flow
+        build_params:
+          components:
+            - type: println
+              name: if-node-true
+              run_params:
+                content: if-node-true
+      condition_false:
+        type: println
+        name: if-node-false
+        run_params:
+          content: if-node-false
+  - type: loop
+    name: loop-flow
+    build_params:
+      condition:
+        type: bool
+        name: condition-node
+        run_params:
+          op: rand
+      sub:
+        type: seq
+        name: sub-flow
+        build_params:
+          components:
+            - type: println
+              name: loop-sub-node
+              run_params:
+                content: loop-sub-node
+  - type: range
+    name: range-flow
+    build_params:
+      values:
+        - range first
+        - range second
+      sub:
+        type: println
+        name: range-sub-node
+        run_params:
+          content: range-sub-node
+  - type: bi
+    name: bi-flow
+    build_params:
+      components:
+        - type: println
+          name: bi-flow-node
+          run_params:
+            content: bi-flow-node
+    run_params:
+      is_bi: true
+      left_params: left
+      right_params: right

pipeline/test/pipeline_test.go

@@ -0,0 +1,41 @@
+package test
+
+import (
+	"fmt"
+	"git.sxidc.com/go-tools/utils/pipeline"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/test/test_node"
+	"testing"
+)
+
+func init() {
+	err := component.RegisterComponentBuilders(
+		&test_node.PrintlnBuilder{}, &test_node.BoolBuilder{},
+	)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func TestPipeline(t *testing.T) {
+	p, err := pipeline.NewPipelineFromYaml("def.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	dynamicParams, err := pipeline.LoadDynamicParamsFromYaml("value.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	globalRunParams := map[string]any{
+		"Test": "Global Params",
+	}
+
+	token := p.Run(globalRunParams, dynamicParams)
+	if token.Wait(); token.Err != nil {
+		t.Fatal(token.Err)
+	}
+
+	fmt.Println(token.Result)
+}

pipeline/test/test_node/bool.go

@@ -0,0 +1,86 @@
+package test_node
+
+import (
+	"errors"
+	"fmt"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/utils"
+	"github.com/mitchellh/mapstructure"
+	"math/rand"
+)
+
+const (
+	TypeBool = "bool"
+)
+
+const (
+	BoolOperationPass          = "pass"
+	BoolOperationReverse       = "reverse"
+	BoolOperationRand          = "rand"
+	BoolOperationUseLastResult = "last"
+)
+
+type BoolRunParams struct {
+	Operation string `mapstructure:"op" structs:"op"`
+	Value     *bool  `mapstructure:"value" structs:"value"`
+}
+
+type BoolNode struct {
+	component.BaseComponent
+}
+
+func (n *BoolNode) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return n.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			params := new(BoolRunParams)
+			err := mapstructure.Decode(dynamicParams, params)
+			if err != nil {
+				return nil, errors.New(TypeBool + "节点运行时参数Decode失败: " + err.Error())
+			}
+
+			if utils.IsStringEmpty(params.Operation) {
+				params.Operation = BoolOperationPass
+			}
+
+			switch params.Operation {
+			case BoolOperationPass:
+				if params.Value == nil {
+					return false, nil
+				}
+
+				return *params.Value, nil
+			case BoolOperationReverse:
+				if params.Value == nil {
+					return false, nil
+				}
+
+				return !*params.Value, nil
+			case BoolOperationRand:
+				result := rand.Uint32()%2 == 0
+				fmt.Println("rand bool is", result)
+				return result, nil
+			case BoolOperationUseLastResult:
+				lastResult := globalRunParams.GetLastResult()
+				lastResultBool, ok := lastResult.(bool)
+				if !ok {
+					return lastResult != nil, nil
+				}
+
+				return lastResultBool, nil
+			default:
+				return nil, errors.New(TypeBool + "不支持的操作: " + params.Operation)
+			}
+		})
+}
+
+type BoolBuilder struct{}
+
+func (builder *BoolBuilder) ProductType() string {
+	return TypeBool
+}
+
+func (builder *BoolBuilder) Build(name string, _ map[string]any, runParams map[string]any) (component.Component, error) {
+	return &BoolNode{
+		BaseComponent: *component.NewBaseComponent(TypeBool, name, runParams),
+	}, nil
+}

pipeline/test/test_node/println.go

@@ -0,0 +1,76 @@
+package test_node
+
+import (
+	"errors"
+	"fmt"
+	"git.sxidc.com/go-tools/utils/pipeline/component"
+	"git.sxidc.com/go-tools/utils/pipeline/component/flow"
+	"github.com/mitchellh/mapstructure"
+)
+
+const (
+	TypePrintln = "println"
+)
+
+type PrintlnRunParams struct {
+	Content string `mapstructure:"content"`
+}
+
+type PrintlnNode struct {
+	component.BaseComponent
+}
+
+func (n *PrintlnNode) Run(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+	return n.OnRun(globalRunParams, dynamicParams,
+		func(globalRunParams *component.GlobalRunParams, dynamicParams map[string]any) (any, error) {
+			fmt.Println("----------PRINTLN----------")
+
+			fmt.Println("Global Params:")
+			for key, value := range globalRunParams.CustomRunParams {
+				fmt.Println(key+":", value)
+			}
+
+			rangeIndex, ok := dynamicParams[flow.RangeIndex]
+			if ok {
+				fmt.Println("Range Params Index:", rangeIndex)
+			}
+
+			rangeOnceValue, ok := dynamicParams[flow.RangeOnceValue]
+			if ok {
+				fmt.Println("Range Params:", rangeOnceValue)
+			}
+
+			biLeftParams, ok := dynamicParams[flow.BiLeft]
+			if ok {
+				fmt.Println("Bi Left Params:", biLeftParams)
+			}
+
+			biRightParams, ok := dynamicParams[flow.BiRight]
+			if ok {
+				fmt.Println("Bi Right Params:", biRightParams)
+			}
+
+			params := new(PrintlnRunParams)
+			err := mapstructure.Decode(dynamicParams, params)
+			if err != nil {
+				return nil, errors.New(TypePrintln + "节点运行时参数Decode失败: " + err.Error())
+			}
+
+			fmt.Println(params.Content)
+			fmt.Println("----------PRINTLN----------")
+
+			return nil, nil
+		})
+}
+
+type PrintlnBuilder struct{}
+
+func (builder *PrintlnBuilder) ProductType() string {
+	return TypePrintln
+}
+
+func (builder *PrintlnBuilder) Build(name string, _ map[string]any, runParams map[string]any) (component.Component, error) {
+	return &PrintlnNode{
+		BaseComponent: *component.NewBaseComponent(TypePrintln, name, runParams),
+	}, nil
+}

pipeline/test/value.yaml

@@ -0,0 +1,29 @@
+seq-flow:
+  seq-flow-node:
+    content: "!!!!seq-flow-node!!!!"
+pipeline-node:
+  content: "!!!!pipeline-node!!!!"
+if-flow:
+  condition_flow:
+    bool:
+      op: pass
+      value: true
+  true-flow:
+    if-node-true:
+      content: "!!!!if-node-true!!!!"
+  if-node-false:
+    content: "!!!!if-node-false!!!!"
+loop-flow:
+  sub-flow:
+    loop-sub-node:
+      content: "!!!!loop-sub-node!!!!"
+range-flow:
+  range-sub-node:
+    content: "!!!!range-sub-node!!!!"
+bi-flow:
+  is_bi: true
+  left_params: "!!!!left!!!!"
+  right_params: "!!!!right!!!!"
+  bi-flow-node:
+    content: "!!!!bi-flow-node!!!!"
+

pipeline/utils/utils.go

@@ -0,0 +1,27 @@
+package utils
+
+import (
+	"os"
+	"strings"
+)
+
+func IsStringEmpty(s string) bool {
+	return strings.Trim(s, " ") == ""
+}
+
+func IsStringNotEmpty(s string) bool {
+	return strings.Trim(s, " ") != ""
+}
+
+func PathExists(path string) bool {
+	_, err := os.Stat(path)
+	if err == nil {
+		return true
+	}
+
+	if os.IsNotExist(err) {
+		return false
+	}
+
+	return false
+}