Merge pull request #425 from usual2970/feat/result_branch

添加执行结果分支节点

internal/workflow/node-processor/execute_failure_node.go

@@ -0,0 +1,27 @@
+package nodeprocessor
+
+import (
+	"context"
+
+	"github.com/usual2970/certimate/internal/domain"
+)
+
+type executeFailureNode struct {
+	node *domain.WorkflowNode
+	*nodeLogger
+}
+
+func NewExecuteFailureNode(node *domain.WorkflowNode) *executeFailureNode {
+	return &executeFailureNode{
+		node:       node,
+		nodeLogger: NewNodeLogger(node),
+	}
+}
+
+func (e *executeFailureNode) Run(ctx context.Context) error {
+	e.AddOutput(ctx,
+		e.node.Name,
+		"进入执行失败分支",
+	)
+	return nil
+}

internal/workflow/node-processor/execute_success_node.go

@@ -0,0 +1,27 @@
+package nodeprocessor
+
+import (
+	"context"
+
+	"github.com/usual2970/certimate/internal/domain"
+)
+
+type executeSuccessNode struct {
+	node *domain.WorkflowNode
+	*nodeLogger
+}
+
+func NewExecuteSuccessNode(node *domain.WorkflowNode) *executeSuccessNode {
+	return &executeSuccessNode{
+		node:       node,
+		nodeLogger: NewNodeLogger(node),
+	}
+}
+
+func (e *executeSuccessNode) Run(ctx context.Context) error {
+	e.AddOutput(ctx,
+		e.node.Name,
+		"进入执行成功分支",
+	)
+	return nil
+}

internal/workflow/node-processor/processor.go

@@ -8,7 +8,7 @@ import (
 	"github.com/usual2970/certimate/internal/domain"
 )
 
-type nodeProcessor interface {
+type NodeProcessor interface {
 	Run(ctx context.Context) error
 	Log(ctx context.Context) *domain.WorkflowRunLog
 	AddOutput(ctx context.Context, title, content string, err ...string)
@@ -58,7 +58,7 @@ func (l *nodeLogger) AddOutput(ctx context.Context, title, content string, err .
 	l.log.Outputs = append(l.log.Outputs, output)
 }
 
-func GetProcessor(node *domain.WorkflowNode) (nodeProcessor, error) {
+func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) {
 	switch node.Type {
 	case domain.WorkflowNodeTypeStart:
 		return NewStartNode(node), nil
@@ -70,6 +70,10 @@ func GetProcessor(node *domain.WorkflowNode) (nodeProcessor, error) {
 		return NewDeployNode(node), nil
 	case domain.WorkflowNodeTypeNotify:
 		return NewNotifyNode(node), nil
+	case domain.WorkflowNodeTypeExecuteSuccess:
+		return NewExecuteSuccessNode(node), nil
+	case domain.WorkflowNodeTypeExecuteFailure:
+		return NewExecuteFailureNode(node), nil
 	}
 	return nil, errors.New("not implemented")
 }

internal/workflow/processor/processor.go

@@ -31,7 +31,7 @@ func (w *workflowProcessor) Run(ctx context.Context) error {
 func (w *workflowProcessor) processNode(ctx context.Context, node *domain.WorkflowNode) error {
 	current := node
 	for current != nil {
-		if current.Type == domain.WorkflowNodeTypeBranch {
+		if current.Type == domain.WorkflowNodeTypeBranch || current.Type == domain.WorkflowNodeTypeExecuteResultBranch {
 			for _, branch := range current.Branches {
 				if err := w.processNode(ctx, &branch); err != nil {
 					continue
@@ -39,24 +39,37 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl
 			}
 		}
 
-		if current.Type != domain.WorkflowNodeTypeBranch {
-			processor, err := nodes.GetProcessor(current)
-			if err != nil {
-				return err
-			}
+		var runErr error
+		var processor nodes.NodeProcessor
+		for {
+			if current.Type != domain.WorkflowNodeTypeBranch && current.Type != domain.WorkflowNodeTypeExecuteResultBranch {
+				processor, runErr = nodes.GetProcessor(current)
+				if runErr != nil {
+					break
+				}
 
-			err = processor.Run(ctx)
+				runErr = processor.Run(ctx)
 
-			log := processor.Log(ctx)
-			if log != nil {
-				w.logs = append(w.logs, *log)
-			}
-
-			if err != nil {
-				return err
+				log := processor.Log(ctx)
+				if log != nil {
+					w.logs = append(w.logs, *log)
+				}
+				if runErr != nil {
+					break
+				}
 			}
+			break
+		}
+
+		if runErr != nil && current.Next != nil && current.Next.Type != domain.WorkflowNodeTypeExecuteResultBranch {
+			return runErr
+		} else if runErr != nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
+			current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteFailure)
+		} else if runErr == nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
+			current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteSuccess)
+		} else {
+			current = current.Next
 		}
-		current = current.Next
 
 	}
 	return nil
@@ -65,3 +78,16 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl
 func setContextWorkflowId(ctx context.Context, id string) context.Context {
 	return context.WithValue(ctx, "workflow_id", id)
 }
+
+func GetWorkflowId(ctx context.Context) string {
+	return ctx.Value("workflow_id").(string)
+}
+
+func getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode {
+	for _, branch := range branches {
+		if branch.Type == nodeType {
+			return &branch
+		}
+	}
+	return nil
+}