utils/pipeline/component/flow/seq.go

package flow

import (
	"git.sxidc.com/go-tools/utils/pipeline/component"
	"github.com/mitchellh/mapstructure"
	"github.com/pkg/errors"
)

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
}