| 402 | func (s *LoopStep) Config() *StepConfig { return s.config } |
| 403 | |
| 404 | func (s *LoopStep) Execute(ctx context.Context, input *StepInput) *stream.Reader[*StepOutput] { |
| 405 | reader, writer := stream.Pipe[*StepOutput](1) |
| 406 | |
| 407 | go func() { |
| 408 | defer writer.Close() |
| 409 | startTime := time.Now() |
| 410 | |
| 411 | var iterations []*StepOutput |
| 412 | var lastOutput *StepOutput |
| 413 | |
| 414 | for i := range s.maxIterations { |
| 415 | loopInput := &StepInput{ |
| 416 | Input: input.Input, |
| 417 | PreviousStepOutputs: input.PreviousStepOutputs, |
| 418 | AdditionalData: input.AdditionalData, |
| 419 | SessionState: input.SessionState, |
| 420 | } |
| 421 | |
| 422 | if lastOutput != nil { |
| 423 | loopInput.PreviousStepContent = lastOutput.Content |
| 424 | } |
| 425 | |
| 426 | var iterOutput *StepOutput |
| 427 | bodyReader := s.body.Execute(ctx, loopInput) |
| 428 | for { |
| 429 | output, err := bodyReader.Recv() |
| 430 | if err != nil { |
| 431 | if errors.Is(err, io.EOF) { |
| 432 | break |
| 433 | } |
| 434 | errorOutput := &StepOutput{ |
| 435 | StepID: s.id, |
| 436 | StepName: s.name, |
| 437 | StepType: StepTypeLoop, |
| 438 | Error: err, |
| 439 | StartTime: startTime, |
| 440 | EndTime: time.Now(), |
| 441 | NestedSteps: iterations, |
| 442 | Metadata: map[string]any{"iterations": i, "max": s.maxIterations}, |
| 443 | } |
| 444 | errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds() |
| 445 | writer.Send(errorOutput, err) |
| 446 | return |
| 447 | } |
| 448 | iterOutput = output |
| 449 | } |
| 450 | |
| 451 | iterations = append(iterations, iterOutput) |
| 452 | lastOutput = iterOutput |
| 453 | |
| 454 | if s.stopCondition != nil && s.stopCondition(iterOutput) { |
| 455 | break |
| 456 | } |
| 457 | |
| 458 | if ctx.Err() != nil { |
| 459 | writer.Send(nil, ctx.Err()) |
| 460 | return |
| 461 | } |