utils/pipeline/component/flow/loop.go

package flow

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

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
}