优化蓝图结点

network/ws_server.go

@@ -3,12 +3,13 @@ package network
 import (
 	"crypto/tls"
 	"errors"
-	"github.com/duanhf2012/origin/v2/log"
-	"github.com/gorilla/websocket"
 	"net"
 	"net/http"
 	"sync"
 	"time"
+
+	"github.com/duanhf2012/origin/v2/log"
+	"github.com/gorilla/websocket"
 )
 
 type WSServer struct {
@@ -16,13 +17,16 @@ type WSServer struct {
 	MaxConnNum      int
 	PendingWriteNum int
 	MaxMsgLen       uint32
-	HTTPTimeout     time.Duration
 	CertFile        string
 	KeyFile         string
 	NewAgent        func(*WSConn) Agent
 	ln              net.Listener
 	handler         *WSHandler
 	messageType     int
+
+	HandshakeTimeout time.Duration
+	ReadTimeout      time.Duration
+	WriteTimeout     time.Duration
 }
 
 type WSHandler struct {
@@ -73,14 +77,14 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	handler.conns[conn] = struct{}{}
 	handler.mutexConns.Unlock()
-	c,ok:=conn.NetConn().(*net.TCPConn)
+	c, ok := conn.NetConn().(*net.TCPConn)
 	if !ok {
-		tlsConn,ok := conn.NetConn().(*tls.Conn)
+		tlsConn, ok := conn.NetConn().(*tls.Conn)
 		if !ok {
 			log.Error("conn error")
 			return
 		}
-		c,ok = tlsConn.NetConn().(*net.TCPConn)
+		c, ok = tlsConn.NetConn().(*net.TCPConn)
 		if !ok {
 			log.Error("conn error")
 			return
@@ -127,10 +131,19 @@ func (server *WSServer) Start() error {
 		server.MaxMsgLen = 4096
 		log.Info("invalid MaxMsgLen", log.Uint32("reset", server.MaxMsgLen))
 	}
-	if server.HTTPTimeout <= 0 {
+	if server.HandshakeTimeout <= 0 {
-		server.HTTPTimeout = 10 * time.Second
+		server.HandshakeTimeout = 15 * time.Second
-		log.Info("invalid HTTPTimeout", log.Duration("reset", server.HTTPTimeout))
+		log.Info("invalid HandshakeTimeout", log.Duration("reset", server.HandshakeTimeout))
 	}
+	if server.ReadTimeout <= 0 {
+		server.ReadTimeout = 15 * time.Second
+		log.Info("invalid ReadTimeout", log.Duration("reset", server.ReadTimeout))
+	}
+	if server.WriteTimeout <= 0 {
+		server.WriteTimeout = 15 * time.Second
+		log.Info("invalid WriteTimeout", log.Duration("reset", server.WriteTimeout))
+	}
+
 	if server.NewAgent == nil {
 		log.Error("NewAgent must not be nil")
 		return errors.New("NewAgent must not be nil")
@@ -159,7 +172,7 @@ func (server *WSServer) Start() error {
 		conns:           make(WebsocketConnSet),
 		messageType:     server.messageType,
 		upgrader: websocket.Upgrader{
-			HandshakeTimeout: server.HTTPTimeout,
+			HandshakeTimeout: server.HandshakeTimeout,
 			CheckOrigin:      func(_ *http.Request) bool { return true },
 		},
 	}
@@ -167,8 +180,8 @@ func (server *WSServer) Start() error {
 	httpServer := &http.Server{
 		Addr:           server.Addr,
 		Handler:        server.handler,
-		ReadTimeout:    server.HTTPTimeout,
+		ReadTimeout:    server.ReadTimeout,
-		WriteTimeout:   server.HTTPTimeout,
+		WriteTimeout:   server.WriteTimeout,
 		MaxHeaderBytes: 1024,
 	}
 

sysmodule/netmodule/wsmodule/WSModule.go

@@ -2,13 +2,15 @@ package wsmodule
 
 import (
 	"fmt"
+	"sync"
+	"time"
+
 	"github.com/duanhf2012/origin/v2/event"
 	"github.com/duanhf2012/origin/v2/log"
 	"github.com/duanhf2012/origin/v2/network"
 	"github.com/duanhf2012/origin/v2/network/processor"
 	"github.com/duanhf2012/origin/v2/service"
 	"go.mongodb.org/mongo-driver/bson/primitive"
-	"sync"
 )
 
 type WSModule struct {
@@ -36,6 +38,10 @@ type WSCfg struct {
 	LittleEndian    bool //是否小端序
 	KeyFile 		string
 	CertFile 		string
+
+	HandshakeTimeoutSecond time.Duration
+	ReadTimeoutSecond time.Duration
+	WriteTimeoutSecond time.Duration
 }
 
 type WSPackType int8
@@ -63,6 +69,9 @@ func (ws *WSModule) OnInit() error {
 	ws.WSServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
 	ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
 	ws.WSServer.Addr = ws.wsCfg.ListenAddr
+	ws.WSServer.HandshakeTimeout = ws.wsCfg.HandshakeTimeoutSecond*time.Second
+	ws.WSServer.ReadTimeout = ws.wsCfg.ReadTimeoutSecond*time.Second
+	ws.WSServer.WriteTimeout = ws.wsCfg.WriteTimeoutSecond*time.Second
 
 	if ws.wsCfg.KeyFile != "" && ws.wsCfg.CertFile != "" {
 		ws.WSServer.KeyFile = ws.wsCfg.KeyFile

util/blueprint/blueprint.go

@@ -12,16 +12,16 @@ type Blueprint struct {
 	graphPool GraphPool
 
 	blueprintModule IBlueprintModule
-	mapGraph       map[int64]IGraph
+	mapGraph        map[int64]IGraph
-	seedID int64
+	seedID          int64
-	cancelTimer func(*uint64)bool
+	cancelTimer     func(*uint64) bool
 }
 
 func (bm *Blueprint) RegExecNode(execNode IExecNode) {
 	bm.execNodes = append(bm.execNodes, execNode)
 }
 
-func (bm *Blueprint) regSysNode(){
+func (bm *Blueprint) regSysNode() {
 	bm.RegExecNode(&AddInt{})
 	bm.RegExecNode(&SubInt{})
 	bm.RegExecNode(&MulInt{})
@@ -32,7 +32,7 @@ func (bm *Blueprint) regSysNode(){
 	bm.RegExecNode(&Entrance_ArrayParam{})
 	bm.RegExecNode(&Entrance_IntParam{})
 	bm.RegExecNode(&Entrance_Timer{})
-	bm.RegExecNode(&Output{})
+	bm.RegExecNode(&DebugOutput{})
 	bm.RegExecNode(&Sequence{})
 	bm.RegExecNode(&Foreach{})
 	bm.RegExecNode(&ForeachIntArray{})
@@ -50,11 +50,12 @@ func (bm *Blueprint) regSysNode(){
 	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 {
+func (bm *Blueprint) Init(execDefFilePath string, graphFilePath string, blueprintModule IBlueprintModule, cancelTimer func(*uint64) bool) error {
 	bm.regSysNode()
 	err := bm.execPool.Load(execDefFilePath)
 	if err != nil {
@@ -74,7 +75,7 @@ func (bm *Blueprint) Init(execDefFilePath string, graphFilePath string, blueprin
 
 	bm.cancelTimer = cancelTimer
 	bm.blueprintModule = blueprintModule
-	bm.mapGraph = make(map[int64]IGraph,128)
+	bm.mapGraph = make(map[int64]IGraph, 128)
 	return nil
 }
 
@@ -82,32 +83,41 @@ func (bm *Blueprint) Create(graphName string) int64 {
 	if graphName == "" {
 		return 0
 	}
-	
+
 	graphID := atomic.AddInt64(&bm.seedID, 1)
-	bm.mapGraph[graphID] = bm.graphPool.Create(graphName, graphID)
+	gr := bm.graphPool.Create(graphName, graphID)
+	if gr == nil {
+		return 0
+	}
+
+	bm.mapGraph[graphID] = gr
 	return graphID
 }
 
-func (bm *Blueprint) TriggerEvent(graphID int64, eventID int64, args ...any) error{
+func (bm *Blueprint) TriggerEvent(graphID int64, eventID int64, args ...any) error {
 	graph := bm.mapGraph[graphID]
 	if graph == nil {
 		return fmt.Errorf("can not find graph:%d", graphID)
 	}
 
-	_,err:= graph.Do(eventID, args...)
+	_, err := graph.Do(eventID, args...)
 	return err
 }
 
-func (bm *Blueprint) Do(graphID int64, entranceID int64, args ...any) (Port_Array,error){
+func (bm *Blueprint) Do(graphID int64, entranceID int64, args ...any) (Port_Array, error) {
 	graph := bm.mapGraph[graphID]
 	if graph == nil {
-		return nil,fmt.Errorf("can not find graph:%d", graphID)
+		return nil, fmt.Errorf("can not find graph:%d", graphID)
 	}
 
 	return graph.Do(entranceID, args...)
 }
 
 func (bm *Blueprint) ReleaseGraph(graphID int64) {
+	if graphID == 0 {
+		return
+	}
+	
 	defer delete(bm.mapGraph, graphID)
 	graph := bm.mapGraph[graphID]
 	if graph == nil {
@@ -117,7 +127,7 @@ func (bm *Blueprint) ReleaseGraph(graphID int64) {
 	graph.Release()
 }
 
-func (bm *Blueprint) CancelTimerId(graphID int64, timerId *uint64) bool{
+func (bm *Blueprint) CancelTimerId(graphID int64, timerId *uint64) bool {
 	tId := *timerId
 	bm.cancelTimer(timerId)
 
@@ -126,11 +136,11 @@ func (bm *Blueprint) CancelTimerId(graphID int64, timerId *uint64) bool{
 		return false
 	}
 
-	gr,ok := graph.(*Graph)
+	gr, ok := graph.(*Graph)
 	if !ok {
 		return false
 	}
 
 	delete(gr.mapTimerID, tId)
 	return true
-}
+}

util/blueprint/exec.go

@@ -32,8 +32,6 @@ type IExecNode interface {
 	Exec() (int, error) // 返回后续执行的Node的Index
 	GetNextExecLen() int
 	getInnerExecNode() IInnerExecNode
-
-	setVariableName(name string) bool
 }
 
 type innerExecNode struct {
@@ -51,7 +49,7 @@ type innerExecNode struct {
 }
 
 type BaseExecNode struct {
-	*innerExecNode
+	*innerExecNode // 内部注册的执行结点
 
 	// 执行时初始化的数据
 	*ExecContext
@@ -108,26 +106,13 @@ func (bc *BaseExecConfig) GetMaxOutPortId() int {
 	return maxPortId
 }
 
-//func (em *innerExecNode) AppendInPort(port ...IPort) {
-//	if len(em.inPort) == 0 {
-//		em.inPortParamStartIndex = -1
-//	}
-//
-//	for i := 0; i < len(port); i++ {
-//		if !port[i].IsPortExec() && em.inPortParamStartIndex < 0 {
-//			em.inPortParamStartIndex = len(em.inPort)
-//		}
-//
-//		em.inPort = append(em.inPort, port[i])
-//	}
-//}
 
 func (em *innerExecNode) PrepareMaxInPortId(maxInPortId int) {
-	em.inPort = make([]IPort, 0, maxInPortId+1)
+	em.inPort = make([]IPort, maxInPortId+1)
 }
 
 func (em *innerExecNode) PrepareMaxOutPortId(maxOutPortId int) {
-	em.outPort = make([]IPort, 0, maxOutPortId+1)
+	em.outPort = make([]IPort, maxOutPortId+1)
 }
 
 func (em *innerExecNode) SetInPortById(id int, port IPort) bool {
@@ -243,6 +228,10 @@ func (em *innerExecNode) GetOutPort(index int) IPort {
 	return em.outPort[index]
 }
 
+func (en *BaseExecNode) GetVariableName() string {
+	return en.execNode.variableName
+}
+
 func (en *BaseExecNode) GetBluePrintModule() IBlueprintModule {
 	return en.gr.IBlueprintModule
 }
@@ -579,9 +568,6 @@ func (en *BaseExecNode) getInnerExecNode() IInnerExecNode {
 	return en.innerExecNode.IExecNode.(IInnerExecNode)
 }
 
-func (en *BaseExecNode) setVariableName(name string) bool {
-	return false
-}
 
 func (en *BaseExecNode) GetBlueprintModule() IBlueprintModule {
 	if en.gr == nil {

util/blueprint/graph.go

@@ -6,10 +6,11 @@ import (
 
 	"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()
@@ -28,6 +29,7 @@ type baseGraph struct {
 }
 
 type Graph struct {
+	graphFileName string
 	graphID int64
 	*baseGraph
 	graphContext
@@ -137,12 +139,18 @@ 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))
+	}
+
 	entranceNode := gr.entrance[entranceID]
 	if entranceNode == nil {
 		return nil, fmt.Errorf("entranceID:%d not found", entranceID)
 	}
 
 	gr.variables = map[string]IPort{}
+	gr.context = map[string]*ExecContext{}
+
 	if gr.globalVariables == nil {
 		gr.globalVariables = map[string]IPort{}
 	}

util/blueprint/graphpool.go

@@ -61,6 +61,7 @@ func (gp *GraphPool) Create(graphName string, graphID int64) IGraph {
 	var graph Graph
 	graph.baseGraph = gr
 	graph.graphID = graphID
+	graph.graphFileName = graphName
 	graph.context = make(map[string]*ExecContext, 4)
 	graph.IBlueprintModule = gp.blueprintModule
 	return &graph
@@ -137,8 +138,6 @@ func (gp *GraphPool) genVarExec(nodeCfg *nodeConfig, graphConfig *graphConfig) (
 	}
 
 	e := gp.execPool.GetExec(nodeName)
-	e.(IExecNode).setVariableName(varName)
-
 	return e, varName
 }
 

util/blueprint/node.go

@@ -2,6 +2,7 @@ package blueprint
 
 import (
 	"fmt"
+	"github.com/duanhf2012/origin/v2/log"
 )
 
 type prePortNode struct {
@@ -102,20 +103,6 @@ func (en *execNode) exec(gr *Graph) (int, error) {
 		return -1, err
 	}
 
-	//defer func() {
-	inPort, outPort := node.GetPorts()
-	debugString := "inPort:"
-	for i := 0; i < len(inPort); i++ {
-		debugString += fmt.Sprintf("%+v,", inPort[i])
-	}
-	debugString += "   outPort:"
-	for i := 0; i < len(outPort); i++ {
-		debugString += fmt.Sprintf("%+v,", outPort[i])
-	}
-
-	fmt.Printf("exec node %s,%s\n", en.execNode.GetName(), debugString)
-	//}()
-
 	return e.Exec()
 }
 
@@ -155,6 +142,10 @@ func (en *execNode) doSetInPort(gr *Graph, index int, inPort IPort) error {
 }
 
 func (en *execNode) Do(gr *Graph, outPortArgs ...any) error {
+	if IsDebug {
+		log.Debug("Start ExecNode", log.String("Name",en.execNode.GetName()))
+	}
+
 	// 重新初始化上下文
 	inPorts, outPorts := en.execNode.CloneInOutPort()
 	gr.context[en.Id] = &ExecContext{
@@ -164,7 +155,7 @@ func (en *execNode) Do(gr *Graph, outPortArgs ...any) error {
 
 	startOutIdx := en.execNode.GetOutPortParamStartIndex()
 	for i := 0; i < len(outPortArgs); i++ {
-		if i >= len(outPorts) {
+		if i+startOutIdx >= len(outPorts) {
 			return fmt.Errorf("args %d not found in node %s", i, en.execNode.GetName())
 		}
 
@@ -194,6 +185,10 @@ func (en *execNode) Do(gr *Graph, outPortArgs ...any) error {
 		return err
 	}
 
+	if IsDebug {
+		log.Debug("End ExecNode", log.String("Name",en.execNode.GetName()),log.Any("InPort",inPorts ),log.Any("OutPort",outPorts))
+	}
+
 	if nextIndex == -1 || en.nextNode == nil {
 		return nil
 	}

util/blueprint/port.go

@@ -213,7 +213,7 @@ func (em *Port[T]) convertInt64(v any) (int64, bool) {
 
 func (em *Port[T]) setAnyVale(v any) error {
 	switch v.(type) {
-	case int, int64:
+	case int8,int16,int32,int, int64,uint8,uint16,uint32,uint, uint64:
 		val, ok := em.convertInt64(v)
 		if !ok {
 			return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)

util/blueprint/register.go

@@ -1,4 +0,0 @@
-package blueprint
-
-
-

util/blueprint/sysnodes.go

@@ -15,8 +15,6 @@ const (
 	EntranceID_Timer      = 3
 )
 
-
-
 type Entrance_ArrayParam struct {
 	BaseExecNode
 }
@@ -53,15 +51,15 @@ func (em *Entrance_Timer) Exec() (int, error) {
 	return 0, nil
 }
 
-type Output struct {
+type DebugOutput struct {
 	BaseExecNode
 }
 
-func (em *Output) GetName() string {
+func (em *DebugOutput) GetName() string {
-	return "Output"
+	return "DebugOutput"
 }
 
-func (em *Output) Exec() (int, error) {
+func (em *DebugOutput) Exec() (int, error) {
 	val, ok := em.GetInPortInt(1)
 	if !ok {
 		return 0, fmt.Errorf("output Exec inParam not found")
@@ -77,7 +75,7 @@ func (em *Output) Exec() (int, error) {
 		return 0, fmt.Errorf("output Exec inParam not found")
 	}
 
-	fmt.Printf("output Exec inParam [%d] [%s] [%v]\n", val, valStr, valArray)
+	log.Debug("DebugOutput Exec", log.Any("param1", val), log.Any("param2", valStr), log.Any("param3", valArray))
 	return 0, nil
 }
 
@@ -119,7 +117,8 @@ func (em *ForeachIntArray) Exec() (int, error) {
 	}
 
 	for i := range array {
-		em.ExecContext.OutputPorts[2].SetInt(array[i].IntVal)
+		em.ExecContext.OutputPorts[2].SetInt(Port_Int(i))
+		em.ExecContext.OutputPorts[3].SetInt(array[i].IntVal)
 		err := em.DoNext(0)
 		if err != nil {
 			return -1, err
@@ -471,6 +470,40 @@ func (em *RangeCompare) Exec() (int, error) {
 	return 0, nil
 }
 
+// EqualSwitch 等于分支==
+type EqualSwitch struct {
+	BaseExecNode
+}
+
+func (em *EqualSwitch) GetName() string {
+	return "EqualSwitch"
+}
+
+func (em *EqualSwitch) Exec() (int, error) {
+	inPortA := em.GetInPort(1)
+	if inPortA == nil {
+		return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
+	}
+
+	ret, ok := inPortA.GetInt()
+	if !ok {
+		return -1, fmt.Errorf("GreaterThanInteger inParam 1 error")
+	}
+
+	intArray := em.execNode.GetInPortDefaultIntArrayValue(2)
+	if intArray == nil {
+		return 0, nil
+	}
+
+	for i := 0; i < len(intArray) && i < em.GetOutPortCount()-2; i++ {
+		if ret == intArray[i] {
+			return i + 2, nil
+		}
+	}
+
+	return 0, nil
+}
+
 // Probability 概率判断(万分比)
 type Probability struct {
 	BaseExecNode

util/blueprint/variables.go

@@ -12,13 +12,11 @@ const globalVariablesPrefix = "g_"
 type GetVariablesNode struct {
 	BaseExecNode
 	nodeName string
-	varName  string
 }
 
 type SetVariablesNode struct {
 	BaseExecNode
 	nodeName string
-	varName  string
 }
 
 func (g *GetVariablesNode) GetName() string {
@@ -27,14 +25,14 @@ func (g *GetVariablesNode) GetName() string {
 
 func (g *GetVariablesNode) Exec() (int, error) {
 	var port IPort
-	if strings.HasPrefix(g.varName, globalVariablesPrefix) {
+	if strings.HasPrefix(g.GetVariableName(), globalVariablesPrefix) {
-		port = g.gr.globalVariables[g.varName]
+		port = g.gr.globalVariables[g.GetVariableName()]
 	} else {
-		port = g.gr.variables[g.varName]
+		port = g.gr.variables[g.GetVariableName()]
 	}
 
 	if port == nil {
-		return -1, fmt.Errorf("variable %s not found,node name %s", g.varName, g.nodeName)
+		return -1, fmt.Errorf("variable %s not found,node name %s", g.GetVariableName(), g.nodeName)
 	}
 
 	if !g.SetOutPort(0, port) {
@@ -44,10 +42,7 @@ func (g *GetVariablesNode) Exec() (int, error) {
 	return 0, nil
 }
 
-func (g *GetVariablesNode) setVariableName(name string) bool {
+
-	g.varName = name
-	return true
-}
 
 func (g *SetVariablesNode) GetName() string {
 	return g.nodeName
@@ -60,10 +55,10 @@ func (g *SetVariablesNode) Exec() (int, error) {
 	}
 
 	varPort := port.Clone()
-	if strings.HasPrefix(g.varName, globalVariablesPrefix) {
+	if strings.HasPrefix(g.GetVariableName(), globalVariablesPrefix) {
-		g.gr.globalVariables[g.varName] = varPort
+		g.gr.globalVariables[g.GetVariableName()] = varPort
 	} else {
-		g.gr.variables[g.varName] = varPort
+		g.gr.variables[g.GetVariableName()] = varPort
 	}
 
 	if !g.SetOutPort(1, varPort) {
@@ -73,7 +68,3 @@ func (g *SetVariablesNode) Exec() (int, error) {
 	return 0, nil
 }
 
-func (g *SetVariablesNode) setVariableName(name string) bool {
-	g.varName = name
-	return true
-}