| 631 | func (s *RouterStep) Config() *StepConfig { return s.config } |
| 632 | |
| 633 | func (s *RouterStep) Execute(ctx context.Context, input *StepInput) *stream.Reader[*StepOutput] { |
| 634 | reader, writer := stream.Pipe[*StepOutput](1) |
| 635 | |
| 636 | go func() { |
| 637 | defer writer.Close() |
| 638 | startTime := time.Now() |
| 639 | |
| 640 | routeName := s.router(input) |
| 641 | step, exists := s.routes[routeName] |
| 642 | if !exists { |
| 643 | if s.defaultStep == nil { |
| 644 | err := fmt.Errorf("route '%s' not found", routeName) |
| 645 | errorOutput := &StepOutput{ |
| 646 | StepID: s.id, |
| 647 | StepName: s.name, |
| 648 | StepType: StepTypeRouter, |
| 649 | Error: err, |
| 650 | StartTime: startTime, |
| 651 | EndTime: time.Now(), |
| 652 | Metadata: map[string]any{"route": routeName}, |
| 653 | } |
| 654 | errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds() |
| 655 | writer.Send(errorOutput, err) |
| 656 | return |
| 657 | } |
| 658 | step = s.defaultStep |
| 659 | routeName = "default" |
| 660 | } |
| 661 | |
| 662 | var routeOutput *StepOutput |
| 663 | stepReader := step.Execute(ctx, input) |
| 664 | for { |
| 665 | output, err := stepReader.Recv() |
| 666 | if err != nil { |
| 667 | if errors.Is(err, io.EOF) { |
| 668 | break |
| 669 | } |
| 670 | errorOutput := &StepOutput{ |
| 671 | StepID: s.id, |
| 672 | StepName: s.name, |
| 673 | StepType: StepTypeRouter, |
| 674 | Error: err, |
| 675 | StartTime: startTime, |
| 676 | EndTime: time.Now(), |
| 677 | Metadata: map[string]any{"route": routeName}, |
| 678 | } |
| 679 | errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds() |
| 680 | writer.Send(errorOutput, err) |
| 681 | return |
| 682 | } |
| 683 | routeOutput = output |
| 684 | } |
| 685 | |
| 686 | output := &StepOutput{ |
| 687 | StepID: s.id, |
| 688 | StepName: s.name, |
| 689 | StepType: StepTypeRouter, |
| 690 | Content: routeOutput.Content, |