utils/pipeline/component/flow/bi.go

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
}