| 516 | func (s *ParallelStep) Config() *StepConfig { return s.config } |
| 517 | |
| 518 | func (s *ParallelStep) Execute(ctx context.Context, input *StepInput) *stream.Reader[*StepOutput] { |
| 519 | reader, writer := stream.Pipe[*StepOutput](1) |
| 520 | |
| 521 | go func() { |
| 522 | defer writer.Close() |
| 523 | startTime := time.Now() |
| 524 | |
| 525 | var wg sync.WaitGroup |
| 526 | results := make([]*StepOutput, len(s.steps)) |
| 527 | errs := make([]error, len(s.steps)) |
| 528 | |
| 529 | for i, step := range s.steps { |
| 530 | wg.Add(1) |
| 531 | go func(index int, st Step) { |
| 532 | defer wg.Done() |
| 533 | stepReader := st.Execute(ctx, input) |
| 534 | for { |
| 535 | output, err := stepReader.Recv() |
| 536 | if err != nil { |
| 537 | if errors.Is(err, io.EOF) { |
| 538 | break |
| 539 | } |
| 540 | errs[index] = err |
| 541 | return |
| 542 | } |
| 543 | results[index] = output |
| 544 | } |
| 545 | }(i, step) |
| 546 | } |
| 547 | |
| 548 | wg.Wait() |
| 549 | |
| 550 | var firstError error |
| 551 | for _, err := range errs { |
| 552 | if err != nil { |
| 553 | firstError = err |
| 554 | break |
| 555 | } |
| 556 | } |
| 557 | |
| 558 | if firstError != nil { |
| 559 | errorOutput := &StepOutput{ |
| 560 | StepID: s.id, |
| 561 | StepName: s.name, |
| 562 | StepType: StepTypeParallel, |
| 563 | Error: firstError, |
| 564 | StartTime: startTime, |
| 565 | EndTime: time.Now(), |
| 566 | NestedSteps: results, |
| 567 | Metadata: map[string]any{"parallel_steps": len(s.steps)}, |
| 568 | } |
| 569 | errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds() |
| 570 | writer.Send(errorOutput, firstError) |
| 571 | return |
| 572 | } |
| 573 | |
| 574 | combinedContent := make(map[string]any) |
| 575 | for i, result := range results { |