Method Execute

pkg/workflow/step.go:309–368

(ctx context.Context, input *StepInput)

func (s *ConditionStep) Config() *StepConfig { return s.config }

func (s *ConditionStep) Execute(ctx context.Context, input *StepInput) *stream.Reader[*StepOutput] {
	reader, writer := stream.Pipe[*StepOutput](1)

	go func() {
		defer writer.Close()
		startTime := time.Now()

		conditionResult := s.condition(input)

		var branch Step
		var branchName string
		if conditionResult {
			branch = s.ifTrue
			branchName = "true"
		} else {
			branch = s.ifFalse
			branchName = "false"
		}

		var branchOutput *StepOutput
		branchReader := branch.Execute(ctx, input)
		for {
			output, err := branchReader.Recv()
			if err != nil {
				if errors.Is(err, io.EOF) {
					break
				}
				errorOutput := &StepOutput{
					StepID:    s.id,
					StepName:  s.name,
					StepType:  StepTypeCondition,
					Error:     err,
					StartTime: startTime,
					EndTime:   time.Now(),
					Metadata:  map[string]any{"condition": conditionResult, "branch": branchName},
				}
				errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds()
				writer.Send(errorOutput, err)
				return
			}
			branchOutput = output
		}

		output := &StepOutput{
			StepID:      s.id,
			StepName:    s.name,
			StepType:    StepTypeCondition,
			Content:     branchOutput.Content,
			StartTime:   startTime,
			EndTime:     time.Now(),
			NestedSteps: []*StepOutput{branchOutput},
			Metadata:    map[string]any{"condition": conditionResult, "branch": branchName},
			Metrics:     &StepMetrics{ExecutionTime: time.Since(startTime).Seconds()},
		}
		output.Duration = output.EndTime.Sub(output.StartTime).Seconds()
		writer.Send(output, nil)
	}()