utils/pipeline/definition.go

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/strutils"
	"github.com/fatih/structs"
)

type Definition struct {
	Name       string                `yaml:"name"`
	Components []ComponentDefinition `yaml:"components"`
}

func (def *Definition) Check() error {
	if strutils.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 strutils.IsStringEmpty(def.Type) {
		return errors.New("组件没有传递类型")
	}

	if strutils.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 strutils.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
}