新增蓝图热更新功能

cluster/cluster.go

@@ -3,13 +3,14 @@ package cluster
 import (
 	"errors"
 	"fmt"
+	"reflect"
+	"strings"
+	"sync"
+
 	"github.com/duanhf2012/origin/v2/event"
 	"github.com/duanhf2012/origin/v2/log"
 	"github.com/duanhf2012/origin/v2/rpc"
 	"github.com/duanhf2012/origin/v2/service"
-	"reflect"
-	"strings"
-	"sync"
 )
 
 var configDir = "./config/"

event/eventtype.go

@@ -10,7 +10,7 @@ const (
 	Sys_Event_Tcp             EventType = -3
 	Sys_Event_Http_Event      EventType = -4
 	Sys_Event_WebSocket       EventType = -5
-	Sys_Event_Kcp         	  EventType = -6
+	Sys_Event_Kcp             EventType = -6
 	Sys_Event_Node_Conn_Event EventType = -7
 	Sys_Event_Nats_Conn_Event EventType = -8
 	Sys_Event_DiscoverService EventType = -9
@@ -18,6 +18,6 @@ const (
 	Sys_Event_EtcdDiscovery   EventType = -11
 	Sys_Event_Gin_Event       EventType = -12
 	Sys_Event_FrameTick       EventType = -13
-
-	Sys_Event_User_Define EventType = 1
+	Sys_Event_ReloadBlueprint EventType = -14
+	Sys_Event_User_Define     EventType = 1
 )

util/blueprint/blueprint.go

@@ -3,79 +3,154 @@ package blueprint
 import (
 	"fmt"
 	"sync/atomic"
+	"github.com/duanhf2012/origin/v2/log"
 )
 
 type Blueprint struct {
 	execNodes []IExecNode // 注册的定义执行结点
-
-	execPool  ExecPool
-	graphPool GraphPool
+	execNodeList []func() IExecNode
+	execPool  *ExecPool
+	graphPool *GraphPool
 
 	blueprintModule IBlueprintModule
 	mapGraph        map[int64]IGraph
 	seedID          int64
 	cancelTimer     func(*uint64) bool
+
+	execDefFilePath string // 执行结点定义文件路径
+	graphFilePath string   // 蓝图文件路径
 }
 
-func (bm *Blueprint) RegExecNode(execNode IExecNode) {
-	bm.execNodes = append(bm.execNodes, execNode)
+func (bm *Blueprint) RegisterExecNode(execNodeFunc func() IExecNode) {
+	bm.execNodeList = append(bm.execNodeList, execNodeFunc)
 }
 
-func (bm *Blueprint) regSysNode() {
-	bm.RegExecNode(&AddInt{})
-	bm.RegExecNode(&SubInt{})
-	bm.RegExecNode(&MulInt{})
-	bm.RegExecNode(&DivInt{})
-	bm.RegExecNode(&ModInt{})
-	bm.RegExecNode(&RandNumber{})
+type IExecNodeType[T any] interface {
+	*T
+	IExecNode
+}
 
-	bm.RegExecNode(&Entrance_ArrayParam{})
-	bm.RegExecNode(&Entrance_IntParam{})
-	bm.RegExecNode(&Entrance_Timer{})
-	bm.RegExecNode(&DebugOutput{})
-	bm.RegExecNode(&Sequence{})
-	bm.RegExecNode(&Foreach{})
-	bm.RegExecNode(&ForeachIntArray{})
+// 生成一个泛型函数，返回func() IExecNode类型
+func  NewExecNode[T any, P IExecNodeType[T]]() func() IExecNode {
+	return func() IExecNode {
+		var t T
+		return P(&t)
+	}
+}
 
-	bm.RegExecNode(&GetArrayInt{})
-	bm.RegExecNode(&GetArrayString{})
-	bm.RegExecNode(&GetArrayLen{})
-	bm.RegExecNode(&CreateIntArray{})
-	bm.RegExecNode(&CreateStringArray{})
-	bm.RegExecNode(&AppendIntegerToArray{})
-	bm.RegExecNode(&AppendStringToArray{})
+func (bm *Blueprint) regSysNodes() {
+	bm.RegisterExecNode(NewExecNode[AddInt]())
+	bm.RegisterExecNode(NewExecNode[SubInt]())
+	bm.RegisterExecNode(NewExecNode[MulInt]())
+	bm.RegisterExecNode(NewExecNode[DivInt]())
+	bm.RegisterExecNode(NewExecNode[ModInt]())
+	bm.RegisterExecNode(NewExecNode[RandNumber]())
+
+	bm.RegisterExecNode(NewExecNode[Entrance_ArrayParam]())
+	bm.RegisterExecNode(NewExecNode[Entrance_IntParam]())
+	bm.RegisterExecNode(NewExecNode[Entrance_Timer]())
+	bm.RegisterExecNode(NewExecNode[DebugOutput]())
+	bm.RegisterExecNode(NewExecNode[Sequence]())
+	bm.RegisterExecNode(NewExecNode[Foreach]())
+	bm.RegisterExecNode(NewExecNode[ForeachIntArray]())
+
+	bm.RegisterExecNode(NewExecNode[GetArrayInt]())
+	bm.RegisterExecNode(NewExecNode[GetArrayString]())
+	bm.RegisterExecNode(NewExecNode[GetArrayLen]())
+	bm.RegisterExecNode(NewExecNode[CreateIntArray]())
+	bm.RegisterExecNode(NewExecNode[CreateStringArray]())
+	bm.RegisterExecNode(NewExecNode[AppendIntegerToArray]())
+	bm.RegisterExecNode(NewExecNode[AppendStringToArray]())
+	bm.RegisterExecNode(NewExecNode[BoolIf]())
+	bm.RegisterExecNode(NewExecNode[GreaterThanInteger]())
+	bm.RegisterExecNode(NewExecNode[LessThanInteger]())
+	bm.RegisterExecNode(NewExecNode[EqualInteger]())
+	bm.RegisterExecNode(NewExecNode[RangeCompare]())
+	bm.RegisterExecNode(NewExecNode[EqualSwitch]())
+	bm.RegisterExecNode(NewExecNode[Probability]())
+	bm.RegisterExecNode(NewExecNode[CreateTimer]())
+}
+
+
+func (bm *Blueprint) StartHotReload() (func(),error) {
+	var execPool ExecPool
+	var graphPool GraphPool
+
+	// 加载配置结点生成名字对应的innerExecNode
+	err := execPool.Load(bm.execDefFilePath)
+	if err != nil {
+		return nil, err
+	}
+
+	// 将注册的实际执行结点与innerExecNode进行关联
+	for _, newExec := range bm.execNodeList {
+		e := newExec()
+		if !execPool.Register(e) {
+			return nil,fmt.Errorf("register exec failed,exec:%s", e.GetName())
+		}
+	}
+
+	// 加载所有的vgf蓝图文件
+	err = graphPool.Load(&execPool, bm.graphFilePath, bm.blueprintModule)
+	if err != nil {
+		return nil, err
+	}
+
+	// 返回配置加载后的刷新内存处理
+	return func() {
+		// 替换旧的执行池和图池
+		bm.execPool = &execPool
+		bm.graphPool = &graphPool
+
+		for _, gh := range bm.mapGraph {
+			gFileName := gh.GetGraphFileName()
+			bGraph := bm.graphPool.GetBaseGraph(gFileName)
+			if bGraph == nil {
+				log.Warn("GetBaseGraph fail", log.String("graph file name", gFileName))
+				bGraph = &baseGraph{entrance: map[int64]*execNode{}}
+			}
+
+			gh.HotReload(bGraph)
+		}
+	}, nil
 
-	bm.RegExecNode(&BoolIf{})
-	bm.RegExecNode(&GreaterThanInteger{})
-	bm.RegExecNode(&LessThanInteger{})
-	bm.RegExecNode(&EqualInteger{})
-	bm.RegExecNode(&RangeCompare{})
-	bm.RegExecNode(&EqualSwitch{})
-	bm.RegExecNode(&Probability{})
-	bm.RegExecNode(&CreateTimer{})
 }
 
 func (bm *Blueprint) Init(execDefFilePath string, graphFilePath string, blueprintModule IBlueprintModule, cancelTimer func(*uint64) bool) error {
-	bm.regSysNode()
-	err := bm.execPool.Load(execDefFilePath)
+	var execPool  ExecPool
+	var graphPool GraphPool
+
+	// 加载配置结点生成名字对应的innerExecNode
+	err := execPool.Load(execDefFilePath)
 	if err != nil {
 		return err
 	}
 
-	for _, e := range bm.execNodes {
-		if !bm.execPool.Register(e) {
+	// 注册系统执行结点
+	bm.regSysNodes()
+
+	// 将注册的实际执行结点与innerExecNode进行关联
+	for _, newExec := range bm.execNodeList {
+		e := newExec()
+		if !execPool.Register(e) {
 			return fmt.Errorf("register exec failed,exec:%s", e.GetName())
 		}
 	}
 
-	err = bm.graphPool.Load(&bm.execPool, graphFilePath, blueprintModule)
+	// 加载所有的vgf蓝图文件
+	err = graphPool.Load(&execPool, graphFilePath, blueprintModule)
 	if err != nil {
 		return err
 	}
 
+	bm.execPool = &execPool
+	bm.graphPool = &graphPool
 	bm.cancelTimer = cancelTimer
 	bm.blueprintModule = blueprintModule
 	bm.mapGraph = make(map[int64]IGraph, 128)
+	bm.execDefFilePath = execDefFilePath
+	bm.graphFilePath = graphFilePath
+
 	return nil
 }
 
@@ -117,7 +192,7 @@ func (bm *Blueprint) ReleaseGraph(graphID int64) {
 	if graphID == 0 {
 		return
 	}
-	
+
 	defer delete(bm.mapGraph, graphID)
 	graph := bm.mapGraph[graphID]
 	if graph == nil {

util/blueprint/blueprint_test.go

@@ -5,6 +5,11 @@ import (
 )
 
 func TestExecMgr(t *testing.T) {
+	var bp Blueprint
+	err := bp.Init("E:\\WorkSpace\\c4\\OriginNodeEditor\\json", "E:\\WorkSpace\\c4\\OriginNodeEditor\\vgf", nil, nil)
+	if err != nil {
+		t.Fatalf("Init failed,err:%v", err)
+	}
 
 	//graphTest2 := bp.Create("testForeach")
 	//err = graphTest2.Do(EntranceID_IntParam, 1, 2, 3)

util/blueprint/exec.go

@@ -2,15 +2,7 @@ package blueprint
 
 import "fmt"
 
-type IBaseExecNode interface {
-	initInnerExecNode(innerNode *innerExecNode)
-	initExecNode(gr *Graph, en *execNode) error
-	GetPorts() ([]IPort, []IPort)
-	getExecNodeInfo() (*ExecContext, *execNode)
-	setExecNodeInfo(gr *ExecContext, en *execNode)
-	GetBlueprintModule() IBlueprintModule
-}
-
+// IInnerExecNode 配置生成的结点
 type IInnerExecNode interface {
 	GetName() string
 	SetExec(exec IExecNode)
@@ -26,7 +18,19 @@ type IInnerExecNode interface {
 	GetOutPortParamStartIndex() int
 }
 
+// IBaseExecNode 实际注册的执行结点的基础结构体
+type IBaseExecNode interface {
+	initInnerExecNode(innerNode *innerExecNode)
+	initExecNode(gr *Graph, en *execNode) error
+	GetPorts() ([]IPort, []IPort)
+	getExecNodeInfo() (*ExecContext, *execNode)
+	setExecNodeInfo(gr *ExecContext, en *execNode)
+	GetBlueprintModule() IBlueprintModule
+}
+
+// IExecNode 实际注册的执行结点
 type IExecNode interface {
+	IBaseExecNode
 	GetName() string
 	DoNext(index int) error
 	Exec() (int, error) // 返回后续执行的Node的Index
@@ -34,6 +38,7 @@ type IExecNode interface {
 	getInnerExecNode() IInnerExecNode
 }
 
+// 配置对应的基础信息+端口数据
 type innerExecNode struct {
 	Name        string
 	Title       string
@@ -45,7 +50,7 @@ type innerExecNode struct {
 
 	outPortParamStartIndex int // 输出参数的起始索引,用于排除执行出口
 
-	IExecNode
+	IExecNode // 实际注册的执行结点
 }
 
 type BaseExecNode struct {
@@ -106,7 +111,6 @@ func (bc *BaseExecConfig) GetMaxOutPortId() int {
 	return maxPortId
 }
 
-
 func (em *innerExecNode) PrepareMaxInPortId(maxInPortId int) {
 	em.inPort = make([]IPort, maxInPortId+1)
 }
@@ -568,7 +572,6 @@ func (en *BaseExecNode) getInnerExecNode() IInnerExecNode {
 	return en.innerExecNode.IExecNode.(IInnerExecNode)
 }
 
-
 func (en *BaseExecNode) GetBlueprintModule() IBlueprintModule {
 	if en.gr == nil {
 		return nil

util/blueprint/execpool.go

@@ -5,19 +5,19 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sort"
 	"strconv"
 	"strings"
-	"sort"
 )
 
-// 格式说明Entrance_ID
+// Entrance 格式:Entrance_XXXX_ID
 const (
 	Entrance = "Entrance_"
 )
 
 type ExecPool struct {
-	innerExecNodeMap map[string]IInnerExecNode
-	execNodeMap      map[string]IExecNode
+	innerExecNodeMap map[string]IInnerExecNode // 所有配置对应的结点信息
+	execNodeMap      map[string]IExecNode      // 实际注册的执行结点
 }
 
 func (em *ExecPool) Load(execDefFilePath string) error {
@@ -49,16 +49,19 @@ func (em *ExecPool) Load(execDefFilePath string) error {
 
 		// 只处理JSON文件
 		if filepath.Ext(path) == ".json" {
+			// 将配置的结点初始化为innerExecNode将加入到innerExecNodeMap中
 			return em.processJSONFile(path)
 		}
 
 		return nil
 	})
+
 	if err != nil {
 		return fmt.Errorf("failed to walk path %s: %v", execDefFilePath, err)
 	}
 
-	return em.loadSysExec()
+	// 生成变量配置对应的配置结点GetVar_类型、SetVar_类型
+	return em.regVariablesNode()
 }
 
 // 处理单个JSON文件
@@ -84,21 +87,24 @@ func (em *ExecPool) processJSONFile(filePath string) error {
 	}
 
 	for i := range baseExecConfig {
-		// 排序
+		// 对PortId进行排序
 		sort.Slice(baseExecConfig[i].Inputs, func(left, right int) bool {
 			return baseExecConfig[i].Inputs[left].PortId < baseExecConfig[i].Inputs[right].PortId
 		})
-
+		// 对PortId进行排序
 		sort.Slice(baseExecConfig[i].Outputs, func(left, right int) bool {
 			return baseExecConfig[i].Outputs[left].PortId < baseExecConfig[i].Outputs[right].PortId
 		})
 
-		exec, err := em.createExecFromJSON(baseExecConfig[i])
-		if err != nil {
-			return err
+		// 根据配置的结点信息，创建innerExecNode
+		var execError error
+		exec, execError := em.createExecFromJSON(baseExecConfig[i])
+		if execError != nil {
+			return execError
 		}
 
-		if !em.loadBaseExec(exec) {
+		// 加载到innerExecNodeMap中
+		if !em.addInnerExec(exec) {
 			return fmt.Errorf("exec %s already registered", exec.GetName())
 		}
 	}
@@ -126,6 +132,7 @@ func (em *ExecPool) createPortByDataType(nodeName, portName, dataType string) (I
 func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExecNode, error) {
 	var baseExec innerExecNode
 
+	// 如果是入口名，则按入口名Entrance_ArrayParam_000002生成结点名:Entrance_ArrayParam
 	entranceName, _, ok := getEntranceNodeNameAndID(baseExecConfig.Name)
 	if ok {
 		baseExec.Name = entranceName
@@ -138,7 +145,7 @@ func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExe
 	baseExec.PrepareMaxInPortId(baseExecConfig.GetMaxInPortId())
 	baseExec.PrepareMaxOutPortId(baseExecConfig.GetMaxOutPortId())
 
-	// exec数量
+	// 初始化所有的输入端口
 	inExecNum := 0
 	for index, input := range baseExecConfig.Inputs {
 		portType := strings.ToLower(input.PortType)
@@ -146,6 +153,7 @@ func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExe
 			return nil, fmt.Errorf("input %s data type %s not support", input.Name, input.DataType)
 		}
 
+		// 输入执行结点只能有一个,且只能放在第一个
 		if portType == Config_PortType_Exec {
 			if inExecNum > 0 {
 				return nil, fmt.Errorf("inPort only allows one Execute,node name %s", baseExec.Name)
@@ -155,19 +163,22 @@ func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExe
 			}
 
 			inExecNum++
+			// 设置执行端口
 			baseExec.SetInPortById(input.PortId, NewPortExec())
-			// baseExec.AppendInPort(NewPortExec())
 			continue
 		}
 
+		// 根据类型设置对应的端口
 		port, err := em.createPortByDataType(baseExec.Name, input.Name, input.DataType)
 		if err != nil {
 			return nil, err
 		}
 
+		// 根据PortId设置端口
 		baseExec.SetInPortById(input.PortId, port)
 	}
 
+	// 初始化所有的输出端口
 	hasData := false
 	for _, output := range baseExecConfig.Outputs {
 		portType := strings.ToLower(output.PortType)
@@ -180,11 +191,13 @@ func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExe
 			return nil, fmt.Errorf("the exec port can only be placed at the front,node name %s", baseExec.Name)
 		}
 
+		// 设置执行端口
 		if portType == Config_PortType_Exec {
 			baseExec.SetOutPortById(output.PortId, NewPortExec())
 			continue
 		}
 
+		// 根据类型设置数据端口
 		hasData = true
 		port, err := em.createPortByDataType(baseExec.Name, output.Name, output.DataType)
 		if err != nil {
@@ -196,7 +209,7 @@ func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExe
 	return &baseExec, nil
 }
 
-func (em *ExecPool) loadBaseExec(exec IInnerExecNode) bool {
+func (em *ExecPool) addInnerExec(exec IInnerExecNode) bool {
 	if _, ok := em.innerExecNodeMap[exec.GetName()]; ok {
 		return false
 	}
@@ -224,10 +237,14 @@ func (em *ExecPool) Register(exec IExecNode) bool {
 		return false
 	}
 
+	// 设置实际执行结点中innerExecNode变量,BaseExecNode.innerExecNode = innerNode
 	baseExecNode.initInnerExecNode(innerNode.(*innerExecNode))
+
+	// innerNode设置实际的exec变量,innerExecNode.IExecNode = exec
 	innerNode.SetExec(exec)
 
-	em.execNodeMap[baseExec.GetName()] = baseExec
+	// 将实际的执行结点保存到execNodeMap中
+	em.execNodeMap[baseExec.GetName()] = exec
 	return true
 }
 
@@ -238,7 +255,8 @@ func (em *ExecPool) GetExec(name string) IInnerExecNode {
 	return nil
 }
 
-func (em *ExecPool) loadSysExec() error {
+// regVariablesNode 注册变量结点GetVar_类型、SetVar_类型
+func (em *ExecPool) regVariablesNode() error {
 	var err error
 	if err = em.regGetVariables(Config_DataType_Int); err != nil {
 		return err
@@ -300,7 +318,7 @@ func (em *ExecPool) regGetVariables(typ string) error {
 	var getVariablesNode GetVariablesNode
 	getVariablesNode.nodeName = baseExec.GetName()
 
-	if !em.loadBaseExec(&baseExec) {
+	if !em.addInnerExec(&baseExec) {
 		return fmt.Errorf("exec %s already registered", baseExec.GetName())
 	}
 	if !em.Register(&getVariablesNode) {
@@ -336,7 +354,7 @@ func (em *ExecPool) regSetVariables(typ string) error {
 	baseExec.SetOutPortById(1, outPort)
 
 	baseExec.IExecNode = &SetVariablesNode{nodeName: baseExec.GetName()}
-	if !em.loadBaseExec(&baseExec) {
+	if !em.addInnerExec(&baseExec) {
 		return fmt.Errorf("exec %s already registered", baseExec.GetName())
 	}
 	if !em.Register(baseExec.IExecNode) {

util/blueprint/graph.go

@@ -4,16 +4,20 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/duanhf2012/origin/v2/log"
 	"github.com/duanhf2012/origin/v2/service"
 	"github.com/goccy/go-json"
-	"github.com/duanhf2012/origin/v2/log"
 )
 
 const ReturnVarial = "g_Return"
+
 var IsDebug = false
+
 type IGraph interface {
 	Do(entranceID int64, args ...any) (Port_Array, error)
 	Release()
+	GetGraphFileName() string
+	HotReload(newBaseGraph *baseGraph)
 }
 
 type IBlueprintModule interface {
@@ -30,7 +34,7 @@ type baseGraph struct {
 
 type Graph struct {
 	graphFileName string
-	graphID int64
+	graphID       int64
 	*baseGraph
 	graphContext
 	IBlueprintModule
@@ -140,7 +144,7 @@ func (gc *graphConfig) GetNodeByID(nodeID string) *nodeConfig {
 
 func (gr *Graph) Do(entranceID int64, args ...any) (Port_Array, error) {
 	if IsDebug {
-		log.Debug("Graph Do", log.String("graphName",gr.graphFileName),log.Int64("graphID", gr.graphID), log.Int64("entranceID", entranceID))
+		log.Debug("Graph Do", log.String("graphName", gr.graphFileName), log.Int64("graphID", gr.graphID), log.Int64("entranceID", entranceID))
 	}
 
 	entranceNode := gr.entrance[entranceID]
@@ -204,3 +208,11 @@ func (gr *Graph) Release() {
 	// 清理掉所有数据
 	*gr = Graph{}
 }
+
+func (gr *Graph) HotReload(newBaseGraph *baseGraph) {
+	gr.baseGraph = newBaseGraph
+}
+
+func (gr *Graph) GetGraphFileName() string{
+	return gr.graphFileName
+}

util/blueprint/graphpool.go

@@ -52,6 +52,15 @@ func (gp *GraphPool) Load(execPool *ExecPool, graphFilePath string, blueprintMod
 	})
 }
 
+func (gp *GraphPool) GetBaseGraph(graphName string) *baseGraph {
+	gr, ok := gp.mapGraphs[graphName]
+	if !ok {
+		return nil
+	}
+
+	return gr
+}
+
 func (gp *GraphPool) Create(graphName string, graphID int64) IGraph {
 	gr, ok := gp.mapGraphs[graphName]
 	if !ok {
@@ -73,21 +82,25 @@ func (gp *GraphPool) processJSONFile(filePath string) error {
 	if err != nil {
 		return fmt.Errorf("failed to open file %s: %v", filePath, err)
 	}
+
 	defer func() {
-		if err := file.Close(); err != nil {
+		if err = file.Close(); err != nil {
 			fmt.Printf("关闭文件 %s 时出错: %v\n", filePath, err)
 		}
 	}()
 
 	fileName := filepath.Base(filePath)
-	ext := filepath.Ext(fileName)             // 获取".html"
+	ext := filepath.Ext(fileName)             // 获取".vgf"
 	name := strings.TrimSuffix(fileName, ext) // 获取"name"
+
+	// 解析文件
 	var gConfig graphConfig
 	decoder := json.NewDecoder(file)
-	if err := decoder.Decode(&gConfig); err != nil {
+	if err = decoder.Decode(&gConfig); err != nil {
 		return fmt.Errorf("failed to decode JSON from file %s: %v", filePath, err)
 	}
 
+	// 预处理蓝图
 	return gp.prepareGraph(name, &gConfig)
 }
 
@@ -173,7 +186,7 @@ func (gp *GraphPool) genAllNode(graphConfig *graphConfig) (map[string]*execNode,
 
 func (gp *GraphPool) prepareOneNode(mapNodeExec map[string]*execNode, nodeExec *execNode, graphConfig *graphConfig, recursion *int) error {
 	*recursion++
-	if *recursion > 100 {
+	if *recursion > 256 {
 		return fmt.Errorf("recursion too deep")
 	}