Compare commits

...

41 Commits

Author SHA1 Message Date
boyce
08effd5bca 添加时间偏移接口 2025-10-16 10:40:47 +08:00
boyce
be91bcd4b5 优化结点 2025-10-14 15:38:08 +08:00
boyce
f22ee230e4 新增时间偏移 2025-10-09 17:22:09 +08:00
boyce
419e7ee0c4 添加模块与定时器功能 2025-10-08 16:16:08 +08:00
boyce
7a34fafdc8 新增数组支持 2025-10-07 22:06:13 +08:00
boyce
640b61bcdb 优化执行结点 2025-10-05 14:56:37 +08:00
boyce
654426a836 优化上下文恢复 2025-10-05 13:37:34 +08:00
boyce
c6488faeff 优化执行结点 2025-10-05 13:20:11 +08:00
boyce
3bf19ed329 新增结点 2025-10-04 21:23:52 +08:00
boyce
d4c0bd22ad 新增结点实现 2025-10-02 11:48:58 +08:00
boyce
6511fc4ac0 优化代码 2025-10-01 22:26:30 +08:00
boyce
be0078015f 新增变量与全局变量 2025-09-24 10:26:19 +08:00
boyce
2a12d40f7a 优化代码 2025-09-23 15:06:57 +08:00
boyce
77e2986ffb 优化代码 2025-09-23 10:20:03 +08:00
boyce
3bcce31a86 优化代码 2025-09-21 18:41:03 +08:00
boyce
a54b3c59fc 新增蓝图执行代码 2025-09-20 07:54:08 +08:00
boyce
6c44ba180c 优化kafkamoudule日志 2025-05-21 21:38:36 +08:00
boyce
ecfd42bdec 优化消息队列日志 2025-05-21 21:34:32 +08:00
boyce
01d3b3e535 优化RankService日志 2025-05-21 21:22:24 +08:00
boyce
550d65a354 优化mysql模块日志 2025-05-21 21:19:03 +08:00
boyce
15580ffce9 新增wss证书配置支持 2025-04-21 21:23:34 +08:00
boyce
bd467a219b 废弃掉HttpService、TcpService、WSService 2025-03-28 10:33:51 +08:00
boyce
af15615345 优化日志生成路径 2025-03-14 18:03:01 +08:00
boyce
50dd80b082 整理代码 2025-03-10 11:35:19 +08:00
boyce
a6487dd41e 将默认日志改为rotatelogs 2025-01-25 00:14:18 +08:00
boyce
d5299294d8 优化日志,新增rotatelogs库支持 2025-01-25 00:04:31 +08:00
boyce
4d36e525a5 优化配置读取,去消默认cluster目录 2025-01-16 13:45:06 +08:00
duanhf2012
3a4350769c 新增etcd认证配置 2025-01-08 18:11:20 +08:00
duanhf2012
d4966ea129 优化ws模块 2024-12-17 14:46:00 +08:00
duanhf2012
3b10eeb792 优化日志 2024-12-16 18:00:26 +08:00
duanhf2012
e3275e9f2a 优化模块释放顺序 2024-12-11 18:31:37 +08:00
duanhf2012
16745b34f0 优化日志 2024-12-11 17:49:59 +08:00
duanhf2012
f34dc7d53f 优化日志自定义Writer 2024-12-11 17:24:06 +08:00
duanhf2012
0a09dc2fee 优化日志 2024-12-11 17:14:29 +08:00
duanhf2012
f01a93c446 优化日志 2024-12-11 17:03:21 +08:00
duanhf2012
4d2ab4ee4f 优化代码 2024-12-11 16:44:09 +08:00
duanhf2012
ffcc5a3489 1.优化服务配置检查
2.废弃SetGoRoutineNum接口
3.释放Module优化
2024-12-06 16:05:25 +08:00
duanhf2012
cf6ca0483b Merge branch 'v2' of https://github.com/duanhf2012/origin into v2 2024-12-05 10:19:24 +08:00
duanhf2012
97a21e6f71 新增Skip接口 2024-12-05 10:19:15 +08:00
duanhf2012
f60a55d03a 优化异步RPC,去掉error返回值 2024-12-04 18:33:53 +08:00
duanhf2012
2c32d6eec9 RPC与日志优化 2024-12-03 17:21:21 +08:00
52 changed files with 3818 additions and 264 deletions

View File

@@ -40,8 +40,6 @@ type NodeInfo struct {
DiscoveryService []DiscoveryService //筛选发现的服务,如果不配置,不进行筛选
status NodeStatus
Retire bool
NetworkName string
}
type NodeRpcInfo struct {

View File

@@ -3,24 +3,23 @@ package cluster
import "github.com/duanhf2012/origin/v2/rpc"
type ConfigDiscovery struct {
funDelNode FunDelNode
funSetNode FunSetNode
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
}
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode FunDelNode,funSetNode FunSetNode) error{
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
discovery.localNodeId = localNodeId
discovery.funDelNode = funDelNode
discovery.funSetNode = funSetNode
//解析本地其他服务配置
_,nodeInfoList,_,err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull)
_, nodeInfoList, _, err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull)
if err != nil {
return err
}
for _,nodeInfo := range nodeInfoList {
for _, nodeInfo := range nodeInfoList {
if nodeInfo.NodeId == localNodeId {
continue
}
@@ -30,5 +29,3 @@ func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode Fu
return nil
}

View File

@@ -5,17 +5,17 @@ import (
"github.com/duanhf2012/origin/v2/service"
)
func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.discoveryInfo.getDiscoveryType() == OriginType { //origin类型服务发现
return cls.setupOriginDiscovery(localNodeId,setupServiceFun)
}else if cls.discoveryInfo.getDiscoveryType() == EtcdType{//etcd类型服务发现
return cls.setupEtcdDiscovery(localNodeId,setupServiceFun)
return cls.setupOriginDiscovery(localNodeId, setupServiceFun)
} else if cls.discoveryInfo.getDiscoveryType() == EtcdType { //etcd类型服务发现
return cls.setupEtcdDiscovery(localNodeId, setupServiceFun)
}
return cls.setupConfigDiscovery(localNodeId,setupServiceFun)
return cls.setupConfigDiscovery(localNodeId, setupServiceFun)
}
func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}
@@ -27,6 +27,7 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
}
cls.serviceDiscovery = getOriginDiscovery()
//2.如果为动态服务发现安装本地发现服务
if localMaster == true {
setupServiceFun(&masterService)
@@ -36,11 +37,10 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
setupServiceFun(&clientService)
cls.AddDiscoveryService(OriginDiscoveryClientName, true)
return nil
}
func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}
@@ -48,12 +48,12 @@ func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun Setup
//setup etcd service
cls.serviceDiscovery = getEtcdDiscovery()
setupServiceFun(cls.serviceDiscovery.(service.IService))
cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(),false)
cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(), false)
return nil
}
func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}

View File

@@ -1,6 +1,11 @@
package cluster
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"github.com/duanhf2012/origin/v2/event"
"github.com/duanhf2012/origin/v2/log"
"github.com/duanhf2012/origin/v2/rpc"
@@ -9,14 +14,11 @@ import (
"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/client/v3"
"google.golang.org/protobuf/proto"
"time"
"context"
"errors"
"fmt"
"go.uber.org/zap"
"os"
"path"
"strings"
"sync/atomic"
"time"
)
const originDir = "/origin"
@@ -40,8 +42,13 @@ type EtcdDiscoveryService struct {
mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
}
var etcdDiscovery *EtcdDiscoveryService
func getEtcdDiscovery() IServiceDiscovery {
etcdDiscovery := &EtcdDiscoveryService{}
if etcdDiscovery == nil {
etcdDiscovery = &EtcdDiscoveryService{}
}
return etcdDiscovery
}
@@ -87,15 +94,43 @@ func (ed *EtcdDiscoveryService) OnInit() error {
}
for i := 0; i < len(etcdDiscoveryCfg.EtcdList); i++ {
client, cerr := clientv3.New(clientv3.Config{
var client *clientv3.Client
var tlsConfig *tls.Config
if etcdDiscoveryCfg.EtcdList[i].Cert != "" {
// load cert
cert, cErr := tls.LoadX509KeyPair(etcdDiscoveryCfg.EtcdList[i].Cert, etcdDiscoveryCfg.EtcdList[i].CertKey)
if cErr != nil {
log.Error("load cert error", log.ErrorField("err", cErr))
return cErr
}
// load root ca
caData, cErr := os.ReadFile(etcdDiscoveryCfg.EtcdList[i].Ca)
if cErr != nil {
log.Error("load root ca error", log.ErrorField("err", cErr))
return cErr
}
pool := x509.NewCertPool()
pool.AppendCertsFromPEM(caData)
tlsConfig = &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: pool,
}
}
client, err = clientv3.New(clientv3.Config{
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
Logger: zap.NewNop(),
Username: etcdDiscoveryCfg.EtcdList[i].UserName,
Password: etcdDiscoveryCfg.EtcdList[i].Password,
Logger: log.GetLogger().Logger,
TLS: tlsConfig,
})
if cerr != nil {
log.Error("etcd discovery init fail", log.ErrorField("err", cerr))
return cerr
if err != nil {
log.Error("etcd discovery init fail", log.ErrorField("err", err))
return err
}
ctx, _ := context.WithTimeout(context.Background(), time.Second*3)

View File

@@ -7,6 +7,7 @@ import (
"github.com/duanhf2012/origin/v2/rpc"
jsoniter "github.com/json-iterator/go"
"gopkg.in/yaml.v3"
"io/fs"
"os"
"path/filepath"
"strings"
@@ -15,9 +16,15 @@ import (
var json = jsoniter.ConfigCompatibleWithStandardLibrary
type EtcdList struct {
NetworkName []string
Endpoints []string
UserName string
Password string
Cert string
CertKey string
Ca string
}
type EtcdDiscovery struct {
@@ -64,12 +71,8 @@ type NodeInfoList struct {
NodeList []NodeInfo
}
func validConfigFile(f os.DirEntry) bool {
if f.IsDir() == true || (filepath.Ext(f.Name()) != ".json" && filepath.Ext(f.Name()) != ".yml" && filepath.Ext(f.Name()) != ".yaml") {
return false
}
return true
func validConfigFile(f string) bool {
return strings.HasSuffix(f, ".json")|| strings.HasSuffix(f, ".yml") || strings.HasSuffix(f, ".yaml")
}
func yamlToJson(data []byte, v interface{}) ([]byte, error) {
@@ -271,32 +274,33 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
var discoveryInfo DiscoveryInfo
var rpcMode RpcMode
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return discoveryInfo, nil, rpcMode, fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _, f := range fileInfoList {
if !validConfigFile(f) {
continue
err := filepath.Walk(configDir, func(path string, info fs.FileInfo, err error)error {
if info.IsDir() {
return nil
}
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
fileNodeInfoList, rErr := cls.ReadClusterConfig(filePath)
if err != nil {
return err
}
if !validConfigFile(info.Name()) {
return nil
}
fileNodeInfoList, rErr := cls.ReadClusterConfig(path)
if rErr != nil {
return discoveryInfo, nil, rpcMode, fmt.Errorf("read file path %s is error:%+v", filePath, rErr)
return fmt.Errorf("read file path %s is error:%+v", path, rErr)
}
err = cls.SetRpcMode(&fileNodeInfoList.RpcMode, &rpcMode)
if err != nil {
return discoveryInfo, nil, rpcMode, err
return err
}
err = discoveryInfo.setDiscovery(&fileNodeInfoList.Discovery)
if err != nil {
return discoveryInfo, nil, rpcMode, err
return err
}
for _, nodeInfo := range fileNodeInfoList.NodeList {
@@ -304,6 +308,12 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
nodeInfoList = append(nodeInfoList, nodeInfo)
}
}
return nil
})
if err != nil {
return discoveryInfo, nil, rpcMode, err
}
if nodeId != rpc.NodeIdNull && (len(nodeInfoList) != 1) {
@@ -325,32 +335,32 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
}
func (cls *Cluster) readLocalService(localNodeId string) error {
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
var globalCfg interface{}
publicService := map[string]interface{}{}
nodeService := map[string]interface{}{}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _, f := range fileInfoList {
if !validConfigFile(f) {
continue
err := filepath.Walk(configDir, func(path string, info fs.FileInfo, err error)error{
if info.IsDir() {
return nil
}
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
if err != nil {
continue
return err
}
if !validConfigFile(info.Name()) {
return nil
}
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(path)
if err != nil {
return err
}
if currGlobalCfg != nil {
//不允许重复的配置global配置
if globalCfg != nil {
return fmt.Errorf("[Global] does not allow repeated configuration in %s", f.Name())
return fmt.Errorf("[Global] does not allow repeated configuration in %s", info.Name())
}
globalCfg = currGlobalCfg
}
@@ -366,7 +376,7 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
pubCfg, ok := serviceConfig[s]
if ok == true {
if _, publicOk := publicService[s]; publicOk == true {
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s", s, f.Name())
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s", s, info.Name())
}
publicService[s] = pubCfg
}
@@ -382,12 +392,17 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
}
if _, nodeOK := nodeService[s]; nodeOK == true {
return fmt.Errorf("NodeService NodeId[%s] Service[%s] does not allow repeated configuration in %s", cls.localNodeInfo.NodeId, s, f.Name())
return fmt.Errorf("NodeService NodeId[%s] Service[%s] does not allow repeated configuration in %s", cls.localNodeInfo.NodeId, s, info.Name())
}
nodeService[s] = nodeCfg
break
}
}
return nil
})
if err != nil {
return err
}
//组合所有的配置
@@ -417,13 +432,12 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
return nil
}
func (cls *Cluster) parseLocalCfg() {
func (cls *Cluster) parseLocalCfg() error{
rpcInfo := NodeRpcInfo{}
rpcInfo.nodeInfo = cls.localNodeInfo
rpcInfo.client = rpc.NewLClient(rpcInfo.nodeInfo.NodeId, &cls.callSet)
cls.mapRpc[cls.localNodeInfo.NodeId] = &rpcInfo
for _, serviceName := range cls.localNodeInfo.ServiceList {
splitServiceName := strings.Split(serviceName, ":")
if len(splitServiceName) == 2 {
@@ -440,8 +454,13 @@ func (cls *Cluster) parseLocalCfg() {
cls.mapServiceNode[serviceName] = make(map[string]struct{})
}
if _,ok:=cls.mapServiceNode[serviceName][cls.localNodeInfo.NodeId];ok {
return fmt.Errorf("duplicate service %s is configured in node %s", serviceName, cls.localNodeInfo.NodeId)
}
cls.mapServiceNode[serviceName][cls.localNodeInfo.NodeId] = struct{}{}
}
return nil
}
func (cls *Cluster) IsNatsMode() bool {
@@ -474,8 +493,7 @@ func (cls *Cluster) InitCfg(localNodeId string) error {
}
//本地配置服务加到全局map信息中
cls.parseLocalCfg()
return nil
return cls.parseLocalCfg()
}
func (cls *Cluster) IsConfigService(serviceName string) bool {

3
go.mod
View File

@@ -6,8 +6,10 @@ toolchain go1.22.7
require (
github.com/IBM/sarama v1.43.3
github.com/duanhf2012/rotatelogs v0.0.0-20250124024205-39765c212d8a
github.com/gin-gonic/gin v1.10.0
github.com/go-sql-driver/mysql v1.6.0
github.com/goccy/go-json v0.10.2
github.com/gomodule/redigo v1.8.8
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.0
@@ -43,7 +45,6 @@ require (
github.com/go-playground/universal-translator v0.18.1 // indirect
github.com/go-playground/validator/v10 v10.20.0 // indirect
github.com/go-stack/stack v1.8.0 // indirect
github.com/goccy/go-json v0.10.2 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/protobuf v1.5.4 // indirect
github.com/golang/snappy v0.0.4 // indirect

2
go.sum
View File

@@ -20,6 +20,8 @@ github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSV
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/duanhf2012/rotatelogs v0.0.0-20250124024205-39765c212d8a h1:BVmZrOSKTg9ry1YjqY6IjVXmBDsFdX/W+pnvO5cPUDc=
github.com/duanhf2012/rotatelogs v0.0.0-20250124024205-39765c212d8a/go.mod h1:S/NNkpdnXps6VXaYVVDFtqQAm/NKayHxxOAhsrFnCgg=
github.com/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA=
github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho=
github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws=

View File

@@ -1,36 +1,53 @@
package log
import (
"github.com/duanhf2012/rotatelogs"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"gopkg.in/natefinch/lumberjack.v2"
"os"
"path/filepath"
"strings"
"time"
)
var isSetLogger bool
var gLogger = NewDefaultLogger()
var LogLevel zapcore.Level
var MaxSize int
var LogPath string
var OpenConsole *bool
var LogChanLen int
type Logger struct {
*zap.Logger
stack bool
OpenConsole *bool
LogPath string
FileName string
LogLevel zapcore.Level
Encoder zapcore.Encoder
LogConfig *lumberjack.Logger
sugaredLogger *zap.SugaredLogger
FileName string
Skip int
Encoder zapcore.Encoder
SugaredLogger *zap.SugaredLogger
CoreList []zapcore.Core
WriteSyncerFun []func() zapcore.WriteSyncer
}
// 设置Logger
func SetLogger(logger *Logger) {
if logger != nil && isSetLogger == false {
if logger != nil {
gLogger = logger
isSetLogger = true
}
}
// 设置ZapLogger
func SetZapLogger(zapLogger *zap.Logger) {
if zapLogger != nil {
gLogger = &Logger{}
gLogger.Logger = zapLogger
isSetLogger = true
}
}
func GetLogger() *Logger {
return gLogger
}
@@ -39,10 +56,14 @@ func (logger *Logger) SetEncoder(encoder zapcore.Encoder) {
logger.Encoder = encoder
}
func (logger *Logger) SetSkip(skip int) {
logger.Skip = skip
}
func GetJsonEncoder() zapcore.Encoder {
encoderConfig := zap.NewProductionEncoderConfig()
encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
encoderConfig.EncodeCaller = zapcore.ShortCallerEncoder
encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
enc.AppendString(t.Format("2006-01-02 15:04:05.000"))
}
@@ -61,51 +82,105 @@ func GetTxtEncoder() zapcore.Encoder {
return zapcore.NewConsoleEncoder(encoderConfig)
}
func getLogConfig() *lumberjack.Logger {
func (logger *Logger) getLogConfig() *lumberjack.Logger {
return &lumberjack.Logger{
Filename: "",
MaxSize: 2048,
Filename: filepath.Join(LogPath, logger.FileName),
MaxSize: MaxSize,
MaxBackups: 0,
MaxAge: 0,
Compress: false,
LocalTime: true,
}
}
func NewDefaultLogger() *Logger {
logger := Logger{}
logger.Encoder = GetJsonEncoder()
logger.LogConfig = getLogConfig()
logger.LogConfig.LocalTime = true
core := zapcore.NewCore(logger.Encoder, zapcore.AddSync(os.Stdout), zap.InfoLevel)
logger.Logger = zap.New(core, zap.AddCaller(), zap.AddCallerSkip(1))
logger.Init()
return &logger
}
func (logger *Logger) SetLogLevel(level zapcore.Level) {
logger.LogLevel = level
func (logger *Logger) SetSyncers(syncers ...func() zapcore.WriteSyncer) {
logger.WriteSyncerFun = syncers
}
func (logger *Logger) AppendSyncerFun(syncerFun func() zapcore.WriteSyncer) {
logger.WriteSyncerFun = append(logger.WriteSyncerFun, syncerFun)
}
func SetLogLevel(level zapcore.Level) {
LogLevel = level
}
func (logger *Logger) Enabled(zapcore.Level) bool {
return logger.stack
}
func (logger *Logger) Init() {
var coreList []zapcore.Core
func (logger *Logger) NewLumberjackWriter() zapcore.WriteSyncer {
return zapcore.AddSync(
&lumberjack.Logger{
Filename: filepath.Join(LogPath, logger.FileName),
MaxSize: MaxSize,
MaxBackups: 0,
MaxAge: 0,
Compress: false,
LocalTime: true,
})
}
if logger.OpenConsole == nil || *logger.OpenConsole {
core := zapcore.NewCore(logger.Encoder, zapcore.AddSync(os.Stdout), logger.LogLevel)
func (logger *Logger) NewRotatelogsWriter() zapcore.WriteSyncer {
var options []rotatelogs.Option
if MaxSize > 0 {
options = append(options, rotatelogs.WithRotateMaxSize(int64(MaxSize)))
}
if LogChanLen > 0 {
options = append(options, rotatelogs.WithChannelLen(LogChanLen))
}
options = append(options, rotatelogs.WithRotationTime(time.Hour*24))
fileName := strings.TrimRight(logger.FileName, filepath.Ext(logger.FileName))
rotateLogs, err := rotatelogs.NewRotateLogs(LogPath, "20060102/"+fileName+"_20060102_150405", options...)
if err != nil {
panic(err)
}
return zapcore.AddSync(rotateLogs)
}
func (logger *Logger) Init() {
if isSetLogger {
return
}
var syncerList []zapcore.WriteSyncer
if logger.WriteSyncerFun == nil {
syncerList = append(syncerList, logger.NewRotatelogsWriter())
} else {
for _, syncer := range logger.WriteSyncerFun {
syncerList = append(syncerList, syncer())
}
}
var coreList []zapcore.Core
if OpenConsole == nil || *OpenConsole {
syncerList = append(syncerList, zapcore.AddSync(os.Stdout))
}
for _, writer := range syncerList {
core := zapcore.NewCore(logger.Encoder, writer, LogLevel)
coreList = append(coreList, core)
}
if logger.LogPath != "" {
writeSyncer := zapcore.AddSync(logger.LogConfig)
core := zapcore.NewCore(logger.Encoder, writeSyncer, logger.LogLevel)
coreList = append(coreList, core)
if logger.CoreList != nil {
coreList = append(coreList, logger.CoreList...)
}
core := zapcore.NewTee(coreList...)
logger.Logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(logger), zap.AddCallerSkip(1))
logger.sugaredLogger = logger.Logger.Sugar()
logger.Logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(logger), zap.AddCallerSkip(1+logger.Skip))
logger.SugaredLogger = logger.Logger.Sugar()
}
func (logger *Logger) Debug(msg string, fields ...zap.Field) {
@@ -164,85 +239,85 @@ func Fatal(msg string, fields ...zap.Field) {
gLogger.stack = false
}
func Debugf(msg string, args ...any) {
gLogger.sugaredLogger.Debugf(msg, args...)
func Debugf(template string, args ...any) {
gLogger.SugaredLogger.Debugf(template, args...)
}
func Infof(msg string, args ...any) {
gLogger.sugaredLogger.Infof(msg, args...)
func Infof(template string, args ...any) {
gLogger.SugaredLogger.Infof(template, args...)
}
func Warnf(msg string, args ...any) {
gLogger.sugaredLogger.Warnf(msg, args...)
func Warnf(template string, args ...any) {
gLogger.SugaredLogger.Warnf(template, args...)
}
func Errorf(msg string, args ...any) {
gLogger.sugaredLogger.Errorf(msg, args...)
func Errorf(template string, args ...any) {
gLogger.SugaredLogger.Errorf(template, args...)
}
func StackErrorf(msg string, args ...any) {
func StackErrorf(template string, args ...any) {
gLogger.stack = true
gLogger.sugaredLogger.Errorf(msg, args...)
gLogger.SugaredLogger.Errorf(template, args...)
gLogger.stack = false
}
func Fatalf(msg string, args ...any) {
gLogger.sugaredLogger.Fatalf(msg, args...)
func Fatalf(template string, args ...any) {
gLogger.SugaredLogger.Fatalf(template, args...)
}
func (logger *Logger) SDebug(args ...interface{}) {
logger.sugaredLogger.Debugln(args...)
logger.SugaredLogger.Debugln(args...)
}
func (logger *Logger) SInfo(args ...interface{}) {
logger.sugaredLogger.Infoln(args...)
logger.SugaredLogger.Infoln(args...)
}
func (logger *Logger) SWarn(args ...interface{}) {
logger.sugaredLogger.Warnln(args...)
logger.SugaredLogger.Warnln(args...)
}
func (logger *Logger) SError(args ...interface{}) {
logger.sugaredLogger.Errorln(args...)
logger.SugaredLogger.Errorln(args...)
}
func (logger *Logger) SStackError(args ...interface{}) {
gLogger.stack = true
logger.sugaredLogger.Errorln(args...)
logger.SugaredLogger.Errorln(args...)
gLogger.stack = false
}
func (logger *Logger) SFatal(args ...interface{}) {
gLogger.stack = true
logger.sugaredLogger.Fatalln(args...)
logger.SugaredLogger.Fatalln(args...)
gLogger.stack = false
}
func SDebug(args ...interface{}) {
gLogger.sugaredLogger.Debugln(args...)
gLogger.SugaredLogger.Debugln(args...)
}
func SInfo(args ...interface{}) {
gLogger.sugaredLogger.Infoln(args...)
gLogger.SugaredLogger.Infoln(args...)
}
func SWarn(args ...interface{}) {
gLogger.sugaredLogger.Warnln(args...)
gLogger.SugaredLogger.Warnln(args...)
}
func SError(args ...interface{}) {
gLogger.sugaredLogger.Errorln(args...)
gLogger.SugaredLogger.Errorln(args...)
}
func SStackError(args ...interface{}) {
gLogger.stack = true
gLogger.sugaredLogger.Errorln(args...)
gLogger.SugaredLogger.Errorln(args...)
gLogger.stack = false
}
func SFatal(args ...interface{}) {
gLogger.stack = true
gLogger.sugaredLogger.Fatalln(args...)
gLogger.SugaredLogger.Fatalln(args...)
gLogger.stack = false
}

View File

@@ -73,9 +73,22 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
handler.conns[conn] = struct{}{}
handler.mutexConns.Unlock()
c,ok:=conn.NetConn().(*net.TCPConn)
if !ok {
tlsConn,ok := conn.NetConn().(*tls.Conn)
if !ok {
log.Error("conn error")
return
}
c,ok = tlsConn.NetConn().(*net.TCPConn)
if !ok {
log.Error("conn error")
return
}
}
conn.UnderlyingConn().(*net.TCPConn).SetLinger(0)
conn.UnderlyingConn().(*net.TCPConn).SetNoDelay(true)
c.SetLinger(0)
c.SetNoDelay(true)
wsConn := newWSConn(conn, r.Header, handler.pendingWriteNum, handler.maxMsgLen, handler.messageType)
agent := handler.newAgent(wsConn)
agent.Run()

View File

@@ -17,7 +17,6 @@ import (
_ "net/http/pprof"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
@@ -50,7 +49,7 @@ func init() {
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, SingleStop, SignalRetire)
console.RegisterCommandBool("help", false, "<-help> This help.", usage)
console.RegisterCommandString("name", "", "<-name nodeName> Node's name.", setName)
console.RegisterCommandString("name", "", "<-name nodeName> node's name.", setName)
console.RegisterCommandString("start", "", "<-start nodeid=nodeid> Run originserver.", startNode)
console.RegisterCommandString("stop", "", "<-stop nodeid=nodeid> Stop originserver process.", stopNode)
console.RegisterCommandString("retire", "", "<-retire nodeid=nodeid> retire originserver process.", retireNode)
@@ -59,6 +58,7 @@ func init() {
console.RegisterCommandString("loglevel", "debug", "<-loglevel debug|info|warn|error|stackerror|fatal> Set loglevel.", setLevel)
console.RegisterCommandString("logpath", "", "<-logpath path> Set log file path.", setLogPath)
console.RegisterCommandInt("logsize", 0, "<-logsize size> Set log size(MB).", setLogSize)
console.RegisterCommandInt("logchanlen", 0, "<-logchanlen len> Set log channel len.", setLogChanLen)
console.RegisterCommandString("pprof", "", "<-pprof ip:port> Open performance analysis.", setPprof)
}
@@ -220,7 +220,7 @@ func initNode(id string) {
func initLog() error {
logger := log.GetLogger()
if logger.LogPath == "" {
if log.LogPath == "" {
err := setLogPath("./log")
if err != nil {
return err
@@ -230,8 +230,6 @@ func initLog() error {
localNodeInfo := cluster.GetCluster().GetLocalNodeInfo()
fileName := fmt.Sprintf("%s.log", localNodeInfo.NodeId)
logger.FileName = fileName
filepath.Join()
logger.LogConfig.Filename = filepath.Join(logger.LogPath, logger.FileName)
logger.Init()
return nil
@@ -330,13 +328,13 @@ func startNode(args interface{}) error {
myName, mErr := sysprocess.GetMyProcessName()
//当前进程名获取失败,不应该发生
if mErr != nil {
log.Info("get my process's name is error", log.ErrorField("err", mErr))
log.Error("get my process's name is error", log.ErrorField("err", mErr))
os.Exit(-1)
}
//进程id存在而且进程名也相同被认为是当前进程重复运行
if cErr == nil && name == myName {
log.Info("repeat runs are not allowed", log.String("nodeId", strNodeId), log.Int("processId", processId))
log.Error("repeat runs are not allowed", log.String("nodeId", strNodeId), log.Int("processId", processId))
os.Exit(-1)
}
break
@@ -354,10 +352,13 @@ func startNode(args interface{}) error {
service.Start()
//5.运行集群
cluster.GetCluster().Start()
err := cluster.GetCluster().Start()
if err != nil {
log.Error(err.Error())
os.Exit(-1)
}
//6.监听程序退出信号&性能报告
var pProfilerTicker *time.Ticker = &time.Ticker{}
if profilerInterval > 0 {
pProfilerTicker = time.NewTicker(profilerInterval)
@@ -438,10 +439,10 @@ func openConsole(args interface{}) error {
strOpen := strings.ToLower(strings.TrimSpace(args.(string)))
if strOpen == "false" {
bOpenConsole := false
log.GetLogger().OpenConsole = &bOpenConsole
log.OpenConsole = &bOpenConsole
} else if strOpen == "true" {
bOpenConsole := true
log.GetLogger().OpenConsole = &bOpenConsole
log.OpenConsole = &bOpenConsole
} else {
return errors.New("parameter console error")
}
@@ -456,17 +457,17 @@ func setLevel(args interface{}) error {
strlogLevel := strings.TrimSpace(args.(string))
switch strlogLevel {
case "debug":
log.GetLogger().LogLevel = zapcore.DebugLevel
log.LogLevel = zapcore.DebugLevel
case "info":
log.GetLogger().LogLevel = zapcore.InfoLevel
log.LogLevel = zapcore.InfoLevel
case "warn":
log.GetLogger().LogLevel = zapcore.WarnLevel
log.LogLevel = zapcore.WarnLevel
case "error":
log.GetLogger().LogLevel = zapcore.ErrorLevel
log.LogLevel = zapcore.ErrorLevel
case "stackerror":
log.GetLogger().LogLevel = zapcore.ErrorLevel
log.LogLevel = zapcore.ErrorLevel
case "fatal":
log.GetLogger().LogLevel = zapcore.FatalLevel
log.LogLevel = zapcore.FatalLevel
default:
return errors.New("unknown level: " + strlogLevel)
}
@@ -478,19 +479,15 @@ func setLogPath(args interface{}) error {
return nil
}
logPath := strings.TrimSpace(args.(string))
dir, err := os.Stat(logPath)
if err == nil && dir.IsDir() == false {
return errors.New("Not found dir " + logPath)
}
_, err := os.Stat(logPath)
if err != nil {
err = os.Mkdir(log.GetLogger().LogPath, os.ModePerm)
err = os.MkdirAll(logPath, os.ModePerm)
if err != nil {
return errors.New("Cannot create dir " + log.GetLogger().LogPath)
return errors.New("Cannot create dir " + logPath)
}
}
log.GetLogger().LogPath = logPath
log.LogPath = logPath
return nil
}
@@ -503,7 +500,20 @@ func setLogSize(args interface{}) error {
return nil
}
log.GetLogger().LogConfig.MaxSize = logSize
log.MaxSize = logSize
return nil
}
func setLogChanLen(args interface{}) error {
logChanLen, ok := args.(int)
if ok == false {
return errors.New("param logsize is error")
}
if logChanLen == 0 {
return nil
}
log.LogChanLen = logChanLen
return nil
}

View File

@@ -32,7 +32,7 @@ type IRealClient interface {
SetConn(conn *network.NetConn)
Close(waitDone bool)
AsyncCall(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error)
AsyncCall(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error)
Go(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, noReply bool, serviceMethod string, args interface{}, reply interface{}) *Call
RawGo(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, processor IRpcProcessor, noReply bool, rpcMethodId uint32, serviceMethod string, rawArgs []byte, reply interface{}) *Call
IsConnected() bool
@@ -211,7 +211,7 @@ func (client *Client) rawGo(nodeId string, w IWriter, timeout time.Duration, rpc
return call
}
func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
processorType, processor := GetProcessorType(args)
InParam, herr := processor.Marshal(args)
if herr != nil {
@@ -264,10 +264,7 @@ func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration,
return emptyCancelRpc, err
}
if cancelable {
rpcCancel := RpcCancel{CallSeq: seq, Cli: client}
return rpcCancel.CancelRpc, nil
}
return emptyCancelRpc, nil
rpcCancel := RpcCancel{CallSeq: seq, Cli: client}
return rpcCancel.CancelRpc, nil
}

View File

@@ -90,7 +90,7 @@ func (lc *LClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRpcHa
return pLocalRpcServer.selfNodeRpcHandlerGo(timeout, processor, lc.selfClient, true, serviceName, rpcMethodId, serviceName, nil, nil, rawArgs)
}
func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, reply interface{}, cancelable bool) (CancelRpc, error) {
func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, reply interface{}) (CancelRpc, error) {
pLocalRpcServer := rpcHandler.GetRpcServer()()
//判断是否是同一服务
@@ -109,7 +109,7 @@ func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IR
}
//其他的rpcHandler的处理器
cancelRpc, err := pLocalRpcServer.selfNodeRpcHandlerAsyncGo(timeout, lc.selfClient, rpcHandler, false, serviceName, serviceMethod, args, reply, callback, cancelable)
cancelRpc, err := pLocalRpcServer.selfNodeRpcHandlerAsyncGo(timeout, lc.selfClient, rpcHandler, false, serviceName, serviceMethod, args, reply, callback)
if err != nil {
callback.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
}
@@ -121,9 +121,6 @@ func NewLClient(localNodeId string, callSet *CallSet) *Client {
client := &Client{}
client.clientId = atomic.AddUint32(&clientSeq, 1)
client.targetNodeId = localNodeId
//client.maxCheckCallRpcCount = DefaultMaxCheckCallRpcCount
//client.callRpcTimeout = DefaultRpcTimeout
lClient := &LClient{}
lClient.selfClient = client
client.IRealClient = lClient

View File

@@ -127,7 +127,7 @@ func (server *BaseServer) selfNodeRpcHandlerGo(timeout time.Duration, processor
return pCall
}
func (server *BaseServer) selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value, cancelable bool) (CancelRpc, error) {
func (server *BaseServer) selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value) (CancelRpc, error) {
rpcHandler := server.rpcHandleFinder.FindRpcHandler(handlerName)
if rpcHandler == nil {
err := errors.New("service method " + serviceMethod + " not config!")

View File

@@ -63,8 +63,8 @@ func (nc *NatsClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRp
return nc.client.rawGo(nodeId, nc, timeout, rpcHandler, processor, noReply, rpcMethodId, serviceMethod, rawArgs, reply)
}
func (nc *NatsClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
cancelRpc, err := nc.client.asyncCall(nodeId, nc, timeout, rpcHandler, serviceMethod, callback, args, replyParam, cancelable)
func (nc *NatsClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
cancelRpc, err := nc.client.asyncCall(nodeId, nc, timeout, rpcHandler, serviceMethod, callback, args, replyParam)
if err != nil {
callback.Call([]reflect.Value{reflect.ValueOf(replyParam), reflect.ValueOf(err)})
}

View File

@@ -62,8 +62,8 @@ func (rc *RClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRpcHa
return rc.selfClient.rawGo(nodeId, rc, timeout, rpcHandler, processor, noReply, rpcMethodId, serviceMethod, rawArgs, reply)
}
func (rc *RClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
cancelRpc, err := rc.selfClient.asyncCall(nodeId, rc, timeout, rpcHandler, serviceMethod, callback, args, replyParam, cancelable)
func (rc *RClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
cancelRpc, err := rc.selfClient.asyncCall(nodeId, rc, timeout, rpcHandler, serviceMethod, callback, args, replyParam)
if err != nil {
callback.Call([]reflect.Value{reflect.ValueOf(replyParam), reflect.ValueOf(err)})
}

View File

@@ -164,13 +164,6 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
//取出输入参数类型
var rpcMethodInfo RpcMethodInfo
typ := method.Type
if typ.NumOut() != 1 {
return fmt.Errorf("%s The number of returned arguments must be 1", method.Name)
}
if typ.Out(0).String() != "error" {
return fmt.Errorf("%s The return parameter must be of type error", method.Name)
}
if typ.NumIn() < 2 || typ.NumIn() > 4 {
return fmt.Errorf("%s Unsupported parameter format", method.Name)
@@ -183,6 +176,18 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
rpcMethodInfo.hasResponder = true
}
if rpcMethodInfo.hasResponder && typ.NumOut() > 0 {
return fmt.Errorf("%s should not have return parameters", method.Name)
}
if !rpcMethodInfo.hasResponder && typ.NumOut() != 1 {
return fmt.Errorf("%s The number of returned arguments must be 1", method.Name)
}
if !rpcMethodInfo.hasResponder && typ.Out(0).String() != "error" {
return fmt.Errorf("%s The return parameter must be of type error", method.Name)
}
for i := parIdx; i < typ.NumIn(); i++ {
if handler.isExportedOrBuiltinType(typ.In(i)) == false {
return fmt.Errorf("%s Unsupported parameter types", method.Name)
@@ -307,9 +312,11 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
requestHandle := request.requestHandle
returnValues := v.method.Func.Call(paramList)
errInter := returnValues[0].Interface()
if errInter != nil {
err = errInter.(error)
if len(returnValues) > 0 {
errInter := returnValues[0].Interface()
if errInter != nil {
err = errInter.(error)
}
}
if v.hasResponder == false && requestHandle != nil {
@@ -526,8 +533,7 @@ func (handler *RpcHandler) asyncCallRpc(timeout time.Duration, nodeId string, se
}
//2.rpcClient调用
//如果调用本结点服务
return pClientList[0].AsyncCall(pClientList[0].GetTargetNodeId(), timeout, handler.rpcHandler, serviceMethod, fVal, args, reply, false)
return pClientList[0].AsyncCall(pClientList[0].GetTargetNodeId(), timeout, handler.rpcHandler, serviceMethod, fVal, args, reply, )
}
func (handler *RpcHandler) GetName() string {

View File

@@ -27,9 +27,6 @@ func (rn *RpcNats) NewNatsClient(targetNodeId string,localNodeId string,callSet
client.clientId = atomic.AddUint32(&clientSeq, 1)
client.targetNodeId = targetNodeId
//client.maxCheckCallRpcCount = DefaultMaxCheckCallRpcCount
//client.callRpcTimeout = DefaultRpcTimeout
natsClient := &rn.NatsClient
natsClient.localNodeId = localNodeId
natsClient.client = &client

View File

@@ -31,7 +31,7 @@ type IServer interface {
selfNodeRpcHandlerGo(timeout time.Duration, processor IRpcProcessor, client *Client, noReply bool, handlerName string, rpcMethodId uint32, serviceMethod string, args interface{}, reply interface{}, rawArgs []byte) *Call
myselfRpcHandlerGo(client *Client, handlerName string, serviceMethod string, args interface{}, callBack reflect.Value, reply interface{}) error
selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value, cancelable bool) (CancelRpc, error)
selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value) (CancelRpc, error)
}
type writeResponse func(processor IRpcProcessor, connTag string, serviceMethod string, seq uint64, reply interface{}, rpcError RpcError)

View File

@@ -11,6 +11,7 @@ import (
"github.com/duanhf2012/origin/v2/log"
rpcHandle "github.com/duanhf2012/origin/v2/rpc"
"github.com/duanhf2012/origin/v2/util/timer"
"slices"
)
const InitModuleId = 1e9
@@ -46,7 +47,7 @@ type Module struct {
moduleName string //模块名称
parent IModule //父亲
self IModule //自己
child map[uint32]IModule //孩子们
child []IModule //孩子们
mapActiveTimer map[timer.ITimer]struct{}
mapActiveIdTimer map[uint64]timer.ITimer
dispatcher *timer.Dispatcher //timer
@@ -93,10 +94,7 @@ func (m *Module) AddModule(module IModule) (uint32, error) {
pAddModule.moduleId = m.NewModuleId()
}
if m.child == nil {
m.child = map[uint32]IModule{}
}
_, ok := m.child[module.GetModuleId()]
_,ok := m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()]
if ok == true {
return 0, fmt.Errorf("exists module id %d", module.GetModuleId())
}
@@ -109,29 +107,33 @@ func (m *Module) AddModule(module IModule) (uint32, error) {
pAddModule.eventHandler = event.NewEventHandler()
pAddModule.eventHandler.Init(m.eventHandler.GetEventProcessor())
pAddModule.IConcurrent = m.IConcurrent
m.child = append(m.child,module)
m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()] = module
err := module.OnInit()
if err != nil {
delete(m.ancestor.getBaseModule().(*Module).descendants, module.GetModuleId())
m.child = m.child[:len(m.child)-1]
log.Error("module OnInit error",log.String("ModuleName",module.GetModuleName()),log.ErrorField("err",err))
return 0, err
}
m.child[module.GetModuleId()] = module
m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()] = module
log.Debug("Add module " + module.GetModuleName() + " completed")
return module.GetModuleId(), nil
}
func (m *Module) ReleaseModule(moduleId uint32) {
pModule := m.GetModule(moduleId).getBaseModule().(*Module)
pModule.self.OnRelease()
log.Debug("Release module " + pModule.GetModuleName())
//释放子孙
for id := range pModule.child {
m.ReleaseModule(id)
for i:=len(pModule.child)-1; i>=0; i-- {
m.ReleaseModule(pModule.child[i].GetModuleId())
}
pModule.self.OnRelease()
pModule.GetEventHandler().Destroy()
log.Debug("Release module " + pModule.GetModuleName())
for pTimer := range pModule.mapActiveTimer {
pTimer.Cancel()
}
@@ -140,7 +142,10 @@ func (m *Module) ReleaseModule(moduleId uint32) {
t.Cancel()
}
delete(m.child, moduleId)
m.child = slices.DeleteFunc(m.child, func(module IModule) bool {
return module.GetModuleId() == moduleId
})
delete(m.ancestor.getBaseModule().(*Module).descendants, moduleId)
//清理被删除的Module

View File

@@ -192,7 +192,7 @@ func (s *Service) run() {
break
}
if s.profiler != nil {
analyzer = s.profiler.Push("[Req]" + rpcRequest.RpcRequestData.GetServiceMethod())
analyzer = s.profiler.Push("[RpcReq]" + rpcRequest.RpcRequestData.GetServiceMethod()+"."+strconv.Itoa(int(rpcRequest.RpcRequestData.GetRpcMethodId())))
}
s.GetRpcHandler().HandlerRpcRequest(rpcRequest)
@@ -266,8 +266,10 @@ func (s *Service) Release() {
if atomic.AddInt32(&s.isRelease, -1) == -1 {
s.self.OnRelease()
for i:=len(s.child)-1; i>=0; i-- {
s.ReleaseModule(s.child[i].GetModuleId())
}
}
}
func (s *Service) OnRelease() {
@@ -432,6 +434,7 @@ func (s *Service) SetEventChannelNum(num int) {
}
}
// Deprecated: replace it with the OpenConcurrent function
func (s *Service) SetGoRoutineNum(goroutineNum int32) bool {
//已经开始状态不允许修改协程数量,打开性能分析器不允许开多线程
if s.startStatus == true || s.profiler != nil {

View File

@@ -0,0 +1,74 @@
package blueprintmodule
import (
"fmt"
"github.com/duanhf2012/origin/v2/service"
"github.com/duanhf2012/origin/v2/util/blueprint"
"sync/atomic"
)
type BlueprintModule struct {
service.Module
bp blueprint.Blueprint
execDefFilePath string
graphFilePath string
seedGraphID int64
mapGraph map[int64]blueprint.IGraph
}
func (m *BlueprintModule) Init(execDefFilePath string, graphFilePath string) error {
m.execDefFilePath = execDefFilePath
m.graphFilePath = graphFilePath
m.mapGraph = make(map[int64]blueprint.IGraph, 1024)
return nil
}
func (m *BlueprintModule) OnInit() error {
if m.execDefFilePath == "" || m.graphFilePath == "" {
return fmt.Errorf("execDefFilePath or graphFilePath is empty")
}
m.seedGraphID = 1
return m.bp.Init(m.execDefFilePath, m.graphFilePath, m)
}
func (m *BlueprintModule) CreateGraph(graphName string) int64 {
graphID := atomic.AddInt64(&m.seedGraphID, 1)
graph := m.bp.Create(graphName, graphID)
if graph == nil {
return 0
}
m.mapGraph[graphID] = graph
return graphID
}
func (m *BlueprintModule) GetGraph(graphID int64) (blueprint.IGraph, error) {
graph, ok := m.mapGraph[graphID]
if !ok {
return nil, fmt.Errorf("graph not found,graphID:%d", graphID)
}
return graph, nil
}
func (m *BlueprintModule) Do(graphID int64, entranceID int64, args ...any) error {
graph, err := m.GetGraph(graphID)
if err != nil {
return err
}
return graph.Do(entranceID, args...)
}
func (m *BlueprintModule) TriggerEvent(graphID int64, eventID int64, args ...any) error {
graph, err := m.GetGraph(graphID)
if err != nil {
return err
}
return graph.Do(eventID, args...)
}

View File

@@ -248,7 +248,7 @@ func (ch *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession
select {
case msg := <-claim.Messages():
if msg == nil {
log.SWarning("claim will exit", log.Any("topic", claim.Topic()), log.Any("Partition", claim.Partition()))
log.SWarn("claim will exit", log.Any("topic", claim.Topic()), log.Any("Partition", claim.Partition()))
return nil
}
ch.AppendMsg(session, msg)

View File

@@ -86,7 +86,7 @@ func (p *Producer) asyncRun() {
asyncReturn := sm.Metadata.(*AsyncReturn)
asyncReturn.chanReturn <- asyncReturn
case em := <-p.Errors():
log.Error("async kafkamodule error", log.ErrorAttr("err", em.Err))
log.Error("async kafkamodule error", log.ErrorField("err", em.Err))
if em.Msg.Metadata == nil {
break
}

View File

@@ -95,7 +95,7 @@ func (m *MySQLModule) Begin() (*Tx, error) {
var txDBModule Tx
txDb, err := m.db.Begin()
if err != nil {
log.Error("Begin error:%s", err.Error())
log.Error("Begin error", log.ErrorField("err",err))
return &txDBModule, err
}
txDBModule.slowDuration = m.slowDuration
@@ -155,7 +155,7 @@ func (m *MySQLModule) runPing() {
for {
select {
case <-m.pingCoroutine.pintExit:
log.Error("RunPing stopping %s...", fmt.Sprintf("%T", m))
log.Error("RunPing stopping",log.String("url", m.url),log.String("dbname", m.dbname))
return
case <-m.pingCoroutine.tickerPing.C:
if m.db != nil {
@@ -221,12 +221,12 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
datasetList.blur = true
if checkArgs(args) != nil {
log.Error("CheckArgs is error :%s", strQuery)
log.Error("CheckArgs is error",log.String("sql",strQuery))
return &datasetList, fmt.Errorf("checkArgs is error")
}
if db == nil {
log.Error("cannot connect database:%s", strQuery)
log.Error("cannot connect database",log.String("sql", strQuery))
return &datasetList, fmt.Errorf("cannot connect database")
}
@@ -235,10 +235,10 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
timeFuncPass := time.Since(TimeFuncStart)
if checkSlow(slowDuration, timeFuncPass) {
log.Error("DBModule QueryEx Time %s , Query :%s , args :%+v", timeFuncPass, strQuery, args)
log.Error("Query slow",log.Int64("time_ms",timeFuncPass.Milliseconds()),log.String("sql", strQuery), log.Any("args",args))
}
if err != nil {
log.Error("Query:%s(%v)", strQuery, err)
log.Error("Query error", log.String("sql",strQuery),log.ErrorField("err",err))
if rows != nil {
rows.Close()
}
@@ -278,8 +278,8 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
hasRet := rows.NextResultSet()
if hasRet == false {
if rows.Err() != nil {
log.Error("Query:%s(%+v)", strQuery, rows)
if rowErr :=rows.Err();rowErr != nil {
log.Error("NextResultSet error", log.String("sql",strQuery), log.ErrorField("err",rowErr))
}
break
}
@@ -291,12 +291,12 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
func exec(slowDuration time.Duration, db dbControl, strSql string, args ...interface{}) (*DBResult, error) {
ret := &DBResult{}
if db == nil {
log.Error("cannot connect database:%s", strSql)
log.Error("cannot connect database", log.String("sql",strSql))
return ret, fmt.Errorf("cannot connect database")
}
if checkArgs(args) != nil {
log.Error("CheckArgs is error :%s", strSql)
log.Error("CheckArgs is error", log.String("sql",strSql))
return ret, fmt.Errorf("checkArgs is error")
}
@@ -304,10 +304,10 @@ func exec(slowDuration time.Duration, db dbControl, strSql string, args ...inter
res, err := db.Exec(strSql, args...)
timeFuncPass := time.Since(TimeFuncStart)
if checkSlow(slowDuration, timeFuncPass) {
log.Error("DBModule QueryEx Time %s , Query :%s , args :%+v", timeFuncPass, strSql, args)
log.Error("Exec slow",log.Int64("time_ms",timeFuncPass.Milliseconds()),log.String("sql",strSql),log.Any("args",args) )
}
if err != nil {
log.Error("Exec:%s(%v)", strSql, err)
log.Error("Exec error",log.String("sql",strSql),log.ErrorField("err", err))
return nil, err
}

View File

@@ -0,0 +1 @@
package mysqlmodule

View File

@@ -14,7 +14,7 @@ import (
type WSModule struct {
service.Module
wsServer network.WSServer
WSServer network.WSServer
mapClientLocker sync.RWMutex
mapClient map[string]*WSClient
@@ -34,6 +34,8 @@ type WSCfg struct {
PendingWriteNum int
MaxMsgLen uint32
LittleEndian bool //是否小端序
KeyFile string
CertFile string
}
type WSPackType int8
@@ -57,18 +59,23 @@ func (ws *WSModule) OnInit() error {
return fmt.Errorf("please call the Init function correctly")
}
ws.wsServer.MaxConnNum = ws.wsCfg.MaxConnNum
ws.wsServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
ws.wsServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
ws.wsServer.Addr = ws.wsCfg.ListenAddr
ws.WSServer.MaxConnNum = ws.wsCfg.MaxConnNum
ws.WSServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
ws.WSServer.Addr = ws.wsCfg.ListenAddr
//3.设置解析处理器
if ws.wsCfg.KeyFile != "" && ws.wsCfg.CertFile != "" {
ws.WSServer.KeyFile = ws.wsCfg.KeyFile
ws.WSServer.CertFile = ws.wsCfg.CertFile
}
// 设置解析处理器
ws.process.SetByteOrder(ws.wsCfg.LittleEndian)
ws.mapClient = make(map[string]*WSClient, ws.wsServer.MaxConnNum)
ws.wsServer.NewAgent = ws.NewWSClient
ws.mapClient = make(map[string]*WSClient, ws.WSServer.MaxConnNum)
ws.WSServer.NewAgent = ws.NewWSClient
//4.设置网络事件处理
// 设置网络事件处理
ws.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_WebSocket, ws.GetEventHandler(), ws.wsEventHandler)
return nil
@@ -80,7 +87,7 @@ func (ws *WSModule) Init(wsCfg *WSCfg, process processor.IRawProcessor) {
}
func (ws *WSModule) Start() error {
return ws.wsServer.Start()
return ws.WSServer.Start()
}
func (ws *WSModule) wsEventHandler(ev event.IEvent) {
@@ -197,3 +204,7 @@ func (ws *WSModule) SendRawMsg(clientId string, msg []byte) error {
ws.mapClientLocker.Unlock()
return client.wsConn.WriteMsg(msg)
}
func (ws *WSModule) SetMessageType(messageType int) {
ws.WSServer.SetMessageType(messageType)
}

View File

@@ -83,6 +83,7 @@ type HttpSession struct {
sessionDone chan *HttpSession
}
// Deprecated: replace it with the GinModule
type HttpService struct {
service.Service

View File

@@ -104,11 +104,11 @@ func (cs *CustomerSubscriber) UnSubscribe() {
func (cs *CustomerSubscriber) LoadLastIndex() {
for {
if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription")
log.SInfo("topic ", cs.topic, " out of subscription")
break
}
log.Info("customer ", cs.customerId, " start load last index ")
log.SInfo("customer ", cs.customerId, " start load last index ")
lastIndex, ret := cs.subscriber.dataPersist.LoadCustomerIndex(cs.topic, cs.customerId)
if ret == true {
if lastIndex > 0 {
@@ -116,18 +116,18 @@ func (cs *CustomerSubscriber) LoadLastIndex() {
} else {
//否则直接使用客户端发回来的
}
log.Info("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex)
log.SInfo("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex)
break
}
log.Info("customer ", cs.customerId, " load last index is fail...")
log.SInfo("customer ", cs.customerId, " load last index is fail...")
time.Sleep(5 * time.Second)
}
}
func (cs *CustomerSubscriber) SubscribeRun() {
defer cs.subscriber.queueWait.Done()
log.Info("topic ", cs.topic, " start subscription")
log.SInfo("topic ", cs.topic, " start subscription")
//加载之前的位置
if cs.subscribeMethod == MethodLast {
@@ -136,7 +136,7 @@ func (cs *CustomerSubscriber) SubscribeRun() {
for {
if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription")
log.SInfo("topic ", cs.topic, " out of subscription")
break
}
@@ -146,14 +146,14 @@ func (cs *CustomerSubscriber) SubscribeRun() {
//todo 检测退出
if cs.subscribe() == false {
log.Info("topic ", cs.topic, " out of subscription")
log.SInfo("topic ", cs.topic, " out of subscription")
break
}
}
//删除订阅关系
cs.subscriber.removeCustomer(cs.customerId, cs)
log.Info("topic ", cs.topic, " unsubscription")
log.SInfo("topic ", cs.topic, " unsubscription")
}
func (cs *CustomerSubscriber) subscribe() bool {

View File

@@ -63,7 +63,7 @@ func (ms *MessageQueueService) ReadCfg() error {
maxProcessTopicBacklogNum, ok := mapDBServiceCfg["MaxProcessTopicBacklogNum"]
if ok == false {
ms.maxProcessTopicBacklogNum = DefaultMaxTopicBacklogNum
log.Info("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum)
log.SInfo("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum)
} else {
ms.maxProcessTopicBacklogNum = int32(maxProcessTopicBacklogNum.(float64))
}
@@ -71,7 +71,7 @@ func (ms *MessageQueueService) ReadCfg() error {
memoryQueueLen, ok := mapDBServiceCfg["MemoryQueueLen"]
if ok == false {
ms.memoryQueueLen = DefaultMemoryQueueLen
log.Info("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen)
log.SInfo("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen)
} else {
ms.memoryQueueLen = int32(memoryQueueLen.(float64))
}

View File

@@ -237,7 +237,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
defer cancelAll()
err = cursor.All(ctxAll, &res)
if err != nil {
log.Error("find collect name ", topic, " is error", log.ErrorAttr("err", err))
log.Error("find collect name error",log.String("topic",topic) ,log.ErrorField("err",err))
return nil, false
}
@@ -246,7 +246,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
rawData, errM := bson.Marshal(res[i])
if errM != nil {
if errM != nil {
log.Error("collect name ", topic, " Marshal is error", log.ErrorAttr("err", err))
log.Error("Marshal error",log.String("topic",topic) , log.ErrorField("err", err))
return nil, false
}
continue
@@ -391,7 +391,7 @@ func (mp *MongoPersist) GetIndex(topicData *TopicData) uint64 {
if e.Key == "_id" {
errC, seq := convertToNumber[uint64](e.Value)
if errC != nil {
log.Error("value is error:%s,%+v, ", errC.Error(), e.Value)
log.Error("value is error", log.ErrorField("err",errC), log.Any("val",e.Value))
}
return seq

View File

@@ -56,9 +56,9 @@ func (ss *Subscriber) TopicSubscribe(rpcHandler rpc.IRpcHandler, subScribeType r
}
if ok == true {
log.Info("repeat subscription for customer ", customerId)
log.SInfo("repeat subscription for customer ", customerId)
} else {
log.Info("subscription for customer ", customerId)
log.SInfo("subscription for customer ", customerId)
}
}
@@ -72,7 +72,7 @@ func (ss *Subscriber) UnSubscribe(customerId string) {
customerSubscriber, ok := ss.mapCustomer[customerId]
if ok == false {
log.SWarning("failed to unsubscribe customer " + customerId)
log.SWarn("failed to unsubscribe customer ", customerId)
return
}

View File

@@ -93,7 +93,7 @@ func (tr *TopicRoom) Stop() {
func (tr *TopicRoom) topicRoomRun() {
defer tr.queueWait.Done()
log.Info("topic room ", tr.topic, " is running..")
log.SInfo("topic room ", tr.topic, " is running..")
for {
if atomic.LoadInt32(&tr.isStop) != 0 {
break
@@ -145,5 +145,5 @@ func (tr *TopicRoom) topicRoomRun() {
}
tr.customerLocker.Unlock()
log.Info("topic room ", tr.topic, " is stop")
log.SInfo("topic room ", tr.topic, " is stop")
}

View File

@@ -142,13 +142,13 @@ func (mp *MongoPersist) OnSetupRank(manual bool, rankSkip *RankSkip) error {
return nil
}
log.Info("start load rank ", rankSkip.GetRankName(), " from mongodb.")
log.SInfo("start load rank ", rankSkip.GetRankName(), " from mongodb.")
err := mp.loadFromDB(rankSkip.GetRankID(), rankSkip.GetRankName())
if err != nil {
log.SError("load from db is fail :%s", err.Error())
return err
}
log.Info("finish load rank ", rankSkip.GetRankName(), " from mongodb.")
log.SInfo("finish load rank ", rankSkip.GetRankName(), " from mongodb.")
return nil
}
@@ -296,7 +296,7 @@ func (mp *MongoPersist) saveToDB() {
buf := make([]byte, 4096)
l := runtime.Stack(buf, false)
errString := fmt.Sprint(r)
log.Dump(string(buf[:l]), log.String("error", errString))
log.StackError(string(buf[:l]), log.String("error", errString))
}
}()

View File

@@ -14,6 +14,7 @@ import (
"time"
)
// Deprecated: replace it with the TcpModule
type TcpService struct {
tcpServer network.TCPServer
service.Service

View File

@@ -12,6 +12,7 @@ import (
"sync"
)
// Deprecated: replace it with the WSModule
type WSService struct {
service.Service
wsServer network.WSServer

View File

@@ -0,0 +1,96 @@
package blueprint
import (
"fmt"
"sync/atomic"
)
type Blueprint struct {
execPool ExecPool
graphPool GraphPool
blueprintModule IBlueprintModule
mapGraph map[int64]IGraph
seedID int64
cancelTimer func(*uint64)bool
}
func (bm *Blueprint) Init(execDefFilePath string, graphFilePath string, blueprintModule IBlueprintModule,cancelTimer func(*uint64)bool) error {
err := bm.execPool.Load(execDefFilePath)
if err != nil {
return err
}
for _, e := range execNodes {
if !bm.execPool.Register(e) {
return fmt.Errorf("register exec failed,exec:%s", e.GetName())
}
}
err = bm.graphPool.Load(&bm.execPool, graphFilePath, blueprintModule)
if err != nil {
return err
}
bm.cancelTimer = cancelTimer
bm.blueprintModule = blueprintModule
bm.mapGraph = make(map[int64]IGraph,128)
return nil
}
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)
return graphID
}
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...)
return err
}
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 graph.Do(entranceID, args...)
}
func (bm *Blueprint) ReleaseGraph(graphID int64) {
defer delete(bm.mapGraph, graphID)
graph := bm.mapGraph[graphID]
if graph == nil {
return
}
graph.Release()
}
func (bm *Blueprint) CancelTimerId(graphID int64, timerId *uint64) bool{
tId := *timerId
bm.cancelTimer(timerId)
graph := bm.mapGraph[graphID]
if graph == nil {
return false
}
gr,ok := graph.(*Graph)
if !ok {
return false
}
delete(gr.mapTimerID, tId)
return true
}

View File

@@ -0,0 +1,40 @@
package blueprint
import (
"testing"
)
func TestExecMgr(t *testing.T) {
var bp Blueprint
err := bp.Init("D:\\Develop\\OriginNodeEditor\\json", "D:\\Develop\\OriginNodeEditor\\vgf", nil)
if err != nil {
t.Fatalf("init failed,err:%v", err)
}
graphTest1 := bp.Create("testArrayOperator", 0)
err = graphTest1.Do(EntranceID_IntParam, 20, 1, 3)
if err != nil {
t.Fatalf("Do EntranceID_IntParam failed,err:%v", err)
}
graphTest1.Release()
//graphTest2 := bp.Create("testForeach")
//err = graphTest2.Do(EntranceID_IntParam, 1, 2, 3)
//if err != nil {
// t.Fatalf("Do EntranceID_IntParam failed,err:%v", err)
//}
//graphTest2 := bp.Create("test2")
//
//err = graphTest2.Do(EntranceID_IntParam, 1, 2, 3)
//if err != nil {
// t.Fatalf("Do EntranceID_IntParam failed,err:%v", err)
//}
//graph := bp.Create("test1")
//err = graph.Do(EntranceID_IntParam, 1, 2, 3)
//if err != nil {
// t.Fatalf("do failed,err:%v", err)
//}
//graph.Release()
}

10
util/blueprint/context.go Normal file
View File

@@ -0,0 +1,10 @@
package blueprint
type ExecContext struct {
InputPorts []IPort
OutputPorts []IPort
}
func (ec *ExecContext) Reset() {
*ec = ExecContext{}
}

540
util/blueprint/exec.go Normal file
View File

@@ -0,0 +1,540 @@
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
}
type IInnerExecNode interface {
GetName() string
SetExec(exec IExecNode)
IsInPortExec(index int) bool
IsOutPortExec(index int) bool
GetInPortCount() int
GetOutPortCount() int
CloneInOutPort() ([]IPort, []IPort)
GetInPort(index int) IPort
GetOutPort(index int) IPort
GetInPortParamStartIndex() int
GetOutPortParamStartIndex() int
}
type IExecNode interface {
GetName() string
DoNext(index int) error
Exec() (int, error) // 返回后续执行的Node的Index
GetNextExecLen() int
getInnerExecNode() IInnerExecNode
setVariableName(name string) bool
}
type innerExecNode struct {
Name string
Title string
Package string
Description string
inPort []IPort
outPort []IPort
inPortParamStartIndex int // 输入参数的起始索引,用于排除执行入口
outPortParamStartIndex int // 输出参数的起始索引,用于排除执行出口
IExecNode
}
type BaseExecNode struct {
*innerExecNode
// 执行时初始化的数据
*ExecContext
gr *Graph
execNode *execNode
}
type InputConfig struct {
Name string `json:"name"`
PortType string `json:"type"`
DataType string `json:"data_type"`
HasInput bool `json:"has_input"`
PinWidget string `json:"pin_widget"`
}
type OutInputConfig struct {
Name string `json:"name"`
PortType string `json:"type"`
DataType string `json:"data_type"`
HasInput bool `json:"has_input"`
}
type BaseExecConfig struct {
Name string `json:"name"`
Title string `json:"title"`
Package string `json:"package"`
Description string `json:"description"`
IsPure bool `json:"is_pure"`
Inputs []InputConfig `json:"inputs"`
Outputs []OutInputConfig `json:"outputs"`
}
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) AppendOutPort(port ...IPort) {
if len(em.outPort) == 0 {
em.outPortParamStartIndex = -1
}
for i := 0; i < len(port); i++ {
if !port[i].IsPortExec() && em.outPortParamStartIndex < 0 {
em.outPortParamStartIndex = len(em.outPort)
}
em.outPort = append(em.outPort, port[i])
}
}
func (em *innerExecNode) GetName() string {
return em.Name
}
func (em *innerExecNode) SetExec(exec IExecNode) {
em.IExecNode = exec
}
func (em *innerExecNode) CloneInOutPort() ([]IPort, []IPort) {
inPorts := make([]IPort, 0, 2)
for _, port := range em.inPort {
if port.IsPortExec() {
// 执行入口, 不需要克隆,占位处理
inPorts = append(inPorts, nil)
continue
}
inPorts = append(inPorts, port.Clone())
}
outPorts := make([]IPort, 0, 2)
for _, port := range em.outPort {
if port.IsPortExec() {
outPorts = append(outPorts, nil)
continue
}
outPorts = append(outPorts, port.Clone())
}
return inPorts, outPorts
}
func (em *innerExecNode) IsInPortExec(index int) bool {
if index >= len(em.inPort) || index < 0 {
return false
}
return em.inPort[index].IsPortExec()
}
func (em *innerExecNode) IsOutPortExec(index int) bool {
if index >= len(em.outPort) || index < 0 {
return false
}
return em.outPort[index].IsPortExec()
}
func (em *innerExecNode) GetInPortCount() int {
return len(em.inPort)
}
func (em *innerExecNode) GetOutPortCount() int {
return len(em.outPort)
}
func (em *innerExecNode) GetInPort(index int) IPort {
if index >= len(em.inPort) || index < 0 {
return nil
}
return em.inPort[index]
}
func (em *innerExecNode) GetOutPort(index int) IPort {
if index >= len(em.outPort) || index < 0 {
return nil
}
return em.outPort[index]
}
func (em *innerExecNode) GetInPortParamStartIndex() int {
return em.inPortParamStartIndex
}
func (em *innerExecNode) GetOutPortParamStartIndex() int {
return em.outPortParamStartIndex
}
func (en *BaseExecNode) GetBluePrintModule() IBlueprintModule {
return en.gr.IBlueprintModule
}
func (en *BaseExecNode) initInnerExecNode(innerNode *innerExecNode) {
en.innerExecNode = innerNode
}
func (en *BaseExecNode) getExecNodeInfo() (*ExecContext, *execNode) {
return en.ExecContext, en.execNode
}
func (en *BaseExecNode) setExecNodeInfo(c *ExecContext, e *execNode) {
en.ExecContext = c
en.execNode = e
}
func (en *BaseExecNode) initExecNode(gr *Graph, node *execNode) error {
ctx, ok := gr.context[node.Id]
if !ok {
return fmt.Errorf("node %s not found", node.Id)
}
en.ExecContext = ctx
en.gr = gr
en.execNode = node
return nil
}
func (en *BaseExecNode) GetPorts() ([]IPort, []IPort) {
return en.InputPorts, en.OutputPorts
}
func (en *BaseExecNode) GetInPort(index int) IPort {
if en.InputPorts == nil {
return nil
}
if index >= len(en.InputPorts) || index < 0 {
return nil
}
return en.InputPorts[index]
}
func (en *BaseExecNode) GetOutPort(index int) IPort {
if en.OutputPorts == nil {
return nil
}
if index >= len(en.OutputPorts) || index < 0 {
return nil
}
return en.OutputPorts[index]
}
func (en *BaseExecNode) SetOutPort(index int, val IPort) bool {
if index >= len(en.OutputPorts) || index < 0 {
return false
}
en.OutputPorts[index].SetValue(val)
return true
}
func (en *BaseExecNode) GetInPortInt(index int) (Port_Int, bool) {
port := en.GetInPort(index)
if port == nil {
return 0, false
}
return port.GetInt()
}
func (en *BaseExecNode) GetInPortFloat(index int) (Port_Float, bool) {
port := en.GetInPort(index)
if port == nil {
return 0, false
}
return port.GetFloat()
}
func (en *BaseExecNode) GetInPortStr(index int) (Port_Str, bool) {
port := en.GetInPort(index)
if port == nil {
return "", false
}
return port.GetStr()
}
func (en *BaseExecNode) GetInPortArray(index int) (Port_Array, bool) {
port := en.GetInPort(index)
if port == nil {
return nil, false
}
return port.GetArray()
}
func (en *BaseExecNode) GetInPortArrayValInt(index int, idx int) (Port_Int, bool) {
port := en.GetInPort(index)
if port == nil {
return 0, false
}
return port.GetArrayValInt(idx)
}
func (en *BaseExecNode) GetInPortArrayValStr(idx int) (Port_Str, bool) {
port := en.GetInPort(idx)
if port == nil {
return "", false
}
return port.GetArrayValStr(idx)
}
func (en *BaseExecNode) GetInPortBool(index int) (Port_Bool, bool) {
port := en.GetInPort(index)
if port == nil {
return false, false
}
return port.GetBool()
}
func (en *BaseExecNode) GetOutPortInt(index int) (Port_Int, bool) {
port := en.GetOutPort(index)
if port == nil {
return 0, false
}
return port.GetInt()
}
func (en *BaseExecNode) GetOutPortFloat(index int) (Port_Float, bool) {
port := en.GetOutPort(index)
if port == nil {
return 0, false
}
return port.GetFloat()
}
func (en *BaseExecNode) GetOutPortStr(index int) (Port_Str, bool) {
port := en.GetOutPort(index)
if port == nil {
return "", false
}
return port.GetStr()
}
func (en *BaseExecNode) GetOutPortArrayValInt(index int, idx int) (Port_Int, bool) {
port := en.GetOutPort(index)
if port == nil {
return 0, false
}
return port.GetArrayValInt(idx)
}
func (en *BaseExecNode) GetOutPortArrayValStr(index int, idx int) (Port_Str, bool) {
port := en.GetOutPort(index)
if port == nil {
return "", false
}
return port.GetArrayValStr(idx)
}
func (en *BaseExecNode) GetOutPortBool(index int) (Port_Bool, bool) {
port := en.GetInPort(index)
if port == nil {
return false, false
}
return port.GetBool()
}
func (en *BaseExecNode) SetInPortInt(index int, val Port_Int) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetInt(val)
}
func (en *BaseExecNode) SetInPortFloat(index int, val Port_Float) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetFloat(val)
}
func (en *BaseExecNode) SetInPortStr(index int, val Port_Str) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetStr(val)
}
func (en *BaseExecNode) SetInBool(index int, val Port_Bool) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetBool(val)
}
func (en *BaseExecNode) SetInPortArrayValInt(index int, idx int, val Port_Int) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetArrayValInt(idx, val)
}
func (en *BaseExecNode) SetInPortArrayValStr(index int, idx int, val Port_Str) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.SetArrayValStr(idx, val)
}
func (en *BaseExecNode) AppendInPortArrayValInt(index int, val Port_Int) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.AppendArrayValInt(val)
}
func (en *BaseExecNode) AppendInPortArrayValStr(index int, val Port_Str) bool {
port := en.GetInPort(index)
if port == nil {
return false
}
return port.AppendArrayValStr(val)
}
func (en *BaseExecNode) GetInPortArrayLen(index int) Port_Int {
port := en.GetInPort(index)
if port == nil {
return 0
}
return port.GetArrayLen()
}
func (en *BaseExecNode) SetOutPortInt(index int, val Port_Int) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetInt(val)
}
func (en *BaseExecNode) SetOutPortFloat(index int, val Port_Float) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetFloat(val)
}
func (en *BaseExecNode) SetOutPortStr(index int, val Port_Str) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetStr(val)
}
func (en *BaseExecNode) SetOutPortBool(index int, val Port_Bool) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetBool(val)
}
func (en *BaseExecNode) SetOutPortArrayValInt(index int, idx int, val Port_Int) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetArrayValInt(idx, val)
}
func (en *BaseExecNode) SetOutPortArrayValStr(index int, idx int, val Port_Str) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.SetArrayValStr(idx, val)
}
func (en *BaseExecNode) AppendOutPortArrayValInt(index int, val Port_Int) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.AppendArrayValInt(val)
}
func (en *BaseExecNode) AppendOutPortArrayValStr(index int, val Port_Str) bool {
port := en.GetOutPort(index)
if port == nil {
return false
}
return port.AppendArrayValStr(val)
}
func (en *BaseExecNode) GetOutPortArrayLen(index int) Port_Int {
port := en.GetOutPort(index)
if port == nil {
return 0
}
return port.GetArrayLen()
}
func (en *BaseExecNode) DoNext(index int) error {
// -1 表示中断运行
if index == -1 {
return nil
}
if index < 0 || index >= len(en.execNode.nextNode) {
return fmt.Errorf("next index %d not found", index)
}
if en.execNode.nextNode[index] == nil {
return nil
}
return en.execNode.nextNode[index].Do(en.gr)
}
func (en *BaseExecNode) GetNextExecLen() int {
return len(en.execNode.nextNode)
}
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 {
return nil
}
return en.gr.IBlueprintModule
}

345
util/blueprint/execpool.go Normal file
View File

@@ -0,0 +1,345 @@
package blueprint
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
)
// 格式说明Entrance_ID
const (
Entrance = "Entrance_"
)
type ExecPool struct {
innerExecNodeMap map[string]IInnerExecNode
execNodeMap map[string]IExecNode
}
func (em *ExecPool) Load(execDefFilePath string) error {
em.innerExecNodeMap = make(map[string]IInnerExecNode, 512)
em.execNodeMap = make(map[string]IExecNode, 512)
// 检查路径是否存在
stat, err := os.Stat(execDefFilePath)
if err != nil {
return fmt.Errorf("failed to access path %s: %v", execDefFilePath, err)
}
// 如果是单个文件,直接处理
if !stat.IsDir() {
return fmt.Errorf("%s is not a directory", execDefFilePath)
}
// 遍历目录及其子目录
err = filepath.Walk(execDefFilePath, func(path string, info os.FileInfo, err error) error {
if err != nil {
fmt.Printf("访问路径出错 %s: %v\n", path, err)
return nil // 继续遍历其他文件
}
// 如果是目录,继续遍历
if info.IsDir() {
return nil
}
// 只处理JSON文件
if filepath.Ext(path) == ".json" {
return em.processJSONFile(path)
}
return nil
})
if err != nil {
return fmt.Errorf("failed to walk path %s: %v", execDefFilePath, err)
}
return em.loadSysExec()
}
// 处理单个JSON文件
func (em *ExecPool) processJSONFile(filePath string) error {
// 打开文件
file, err := os.Open(filePath)
if err != nil {
return fmt.Errorf("failed to open file %s: %v", filePath, err)
}
defer func(file *os.File) {
err = file.Close()
if err != nil {
fmt.Printf("failed to close file %s: %v\n", filePath, err)
return
}
}(file)
var baseExecConfig []BaseExecConfig
decoder := json.NewDecoder(file)
if err = decoder.Decode(&baseExecConfig); err != nil {
return fmt.Errorf("failed to decode JSON from file %s: %v", filePath, err)
}
for i := range baseExecConfig {
exec, err := em.createExecFromJSON(baseExecConfig[i])
if err != nil {
return err
}
if !em.loadBaseExec(exec) {
return fmt.Errorf("exec %s already registered", exec.GetName())
}
}
return nil
}
func (em *ExecPool) createPortByDataType(nodeName, portName, dataType string) (IPort, error) {
switch strings.ToLower(dataType) {
case Config_DataType_Int, Config_DataType_Integer:
return NewPortInt(), nil
case Config_DataType_Float:
return NewPortFloat(), nil
case Config_DataType_Str:
return NewPortStr(), nil
case Config_DataType_Boolean, Config_DataType_Bool:
return NewPortBool(), nil
case Config_DataType_Array:
return NewPortArray(), nil
}
return nil, fmt.Errorf("invalid data type %s,node %s port %s", dataType, nodeName, portName)
}
func (em *ExecPool) createExecFromJSON(baseExecConfig BaseExecConfig) (IInnerExecNode, error) {
var baseExec innerExecNode
entranceName, _, ok := getEntranceNodeNameAndID(baseExecConfig.Name)
if ok {
baseExec.Name = entranceName
} else {
baseExec.Name = baseExecConfig.Name
}
baseExec.Title = baseExecConfig.Title
baseExec.Package = baseExecConfig.Package
baseExec.Description = baseExecConfig.Description
// exec数量
inExecNum := 0
for index, input := range baseExecConfig.Inputs {
portType := strings.ToLower(input.PortType)
if portType != Config_PortType_Exec && portType != Config_PortType_Data {
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)
}
if index > 0 {
return nil, fmt.Errorf("the exec port is only allowed to be placed on the first one,node name %s", baseExec.Name)
}
inExecNum++
baseExec.AppendInPort(NewPortExec())
continue
}
port, err := em.createPortByDataType(baseExec.Name, input.Name, input.DataType)
if err != nil {
return nil, err
}
baseExec.AppendInPort(port)
}
hasData := false
for _, output := range baseExecConfig.Outputs {
portType := strings.ToLower(output.PortType)
if portType != Config_PortType_Exec && portType != Config_PortType_Data {
return nil, fmt.Errorf("output %s data type %s not support,node name %s", output.Name, output.DataType, baseExec.Name)
}
// Exec出口只能先Exec再Data不能穿插如果是Data类型但遇到Exec入口则不允许
if hasData && portType == Config_PortType_Exec {
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.AppendOutPort(NewPortExec())
continue
}
hasData = true
port, err := em.createPortByDataType(baseExec.Name, output.Name, output.DataType)
if err != nil {
return nil, err
}
baseExec.AppendOutPort(port)
}
return &baseExec, nil
}
func (em *ExecPool) loadBaseExec(exec IInnerExecNode) bool {
if _, ok := em.innerExecNodeMap[exec.GetName()]; ok {
return false
}
em.innerExecNodeMap[exec.GetName()] = exec
return true
}
func (em *ExecPool) Register(exec IExecNode) bool {
baseExec, ok := exec.(IExecNode)
if !ok {
return false
}
innerNode, ok := em.innerExecNodeMap[baseExec.GetName()]
if !ok {
return false
}
if _, ok := em.execNodeMap[innerNode.GetName()]; ok {
return false
}
baseExecNode, ok := exec.(IBaseExecNode)
if !ok {
return false
}
baseExecNode.initInnerExecNode(innerNode.(*innerExecNode))
innerNode.SetExec(exec)
em.execNodeMap[baseExec.GetName()] = baseExec
return true
}
func (em *ExecPool) GetExec(name string) IInnerExecNode {
if exec, ok := em.execNodeMap[name]; ok {
return exec.getInnerExecNode()
}
return nil
}
func (em *ExecPool) loadSysExec() error {
var err error
if err = em.regGetVariables(Config_DataType_Int); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Integer); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Float); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Str); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Boolean); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Bool); err != nil {
return err
}
if err = em.regGetVariables(Config_DataType_Array); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Int); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Integer); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Float); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Str); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Boolean); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Bool); err != nil {
return err
}
if err = em.regSetVariables(Config_DataType_Array); err != nil {
return err
}
return nil
}
func (em *ExecPool) regGetVariables(typ string) error {
var baseExec innerExecNode
baseExec.Name = genGetVariablesNodeName(typ)
outPort := NewPortByType(typ)
if outPort == nil {
return fmt.Errorf("invalid type %s", typ)
}
baseExec.AppendOutPort(outPort)
var getVariablesNode GetVariablesNode
getVariablesNode.nodeName = baseExec.GetName()
if !em.loadBaseExec(&baseExec) {
return fmt.Errorf("exec %s already registered", baseExec.GetName())
}
if !em.Register(&getVariablesNode) {
return fmt.Errorf("exec %s already registered", baseExec.GetName())
}
return nil
}
func genSetVariablesNodeName(typ string) string {
return fmt.Sprintf("%s_%s", SetVariables, typ)
}
func genGetVariablesNodeName(typ string) string {
return fmt.Sprintf("%s_%s", GetVariables, typ)
}
func (em *ExecPool) regSetVariables(typ string) error {
var baseExec innerExecNode
baseExec.Name = genSetVariablesNodeName(typ)
inExecPort := NewPortByType(Config_PortType_Exec)
inPort := NewPortByType(typ)
outExecPort := NewPortByType(Config_PortType_Exec)
outPort := NewPortByType(typ)
baseExec.AppendInPort(inExecPort, inPort)
baseExec.AppendOutPort(outExecPort, outPort)
baseExec.IExecNode = &SetVariablesNode{nodeName: baseExec.GetName()}
if !em.loadBaseExec(&baseExec) {
return fmt.Errorf("exec %s already registered", baseExec.GetName())
}
if !em.Register(baseExec.IExecNode) {
return fmt.Errorf("exec %s already registered", baseExec.GetName())
}
return nil
}
func getEntranceNodeNameAndID(className string) (string, int64, bool) {
if !strings.HasPrefix(className, Entrance) {
return "", 0, false
}
parts := strings.Split(className, "_")
if len(parts) != 3 {
return "", 0, false
}
entranceID, err := strconv.Atoi(parts[2])
if err != nil {
return "", 0, false
}
return parts[0] + "_" + parts[1], int64(entranceID), true
}

196
util/blueprint/graph.go Normal file
View File

@@ -0,0 +1,196 @@
package blueprint
import (
"fmt"
"github.com/goccy/go-json"
"time"
"github.com/duanhf2012/origin/v2/service"
)
const ReturnVarial = "g_Return"
type IGraph interface {
Do(entranceID int64, args ...any) (Port_Array,error)
Release()
}
type IBlueprintModule interface {
SafeAfterFunc(timerId *uint64, d time.Duration, AdditionData interface{}, cb func(uint64, interface{}))
TriggerEvent(graphID int64, eventID int64, args ...any) error
CancelTimerId(graphID int64,timerId *uint64) bool
GetGameService() service.IService
GetBattleService() service.IService
}
type baseGraph struct {
entrance map[int64]*execNode // 入口
}
type Graph struct {
graphID int64
*baseGraph
graphContext
IBlueprintModule
mapTimerID map[uint64]struct{}
}
type graphContext struct {
context map[string]*ExecContext // 上下文
variables map[string]IPort // 变量
globalVariables map[string]IPort // 全局变量,g_Return,为执行返回值
}
type nodeConfig struct {
Id string `json:"id"`
Class string `json:"class"`
Module string `json:"module"`
//Pos []float64 `json:"pos"`
PortDefault map[string]interface{} `json:"port_defaultv"`
}
type edgeConfig struct {
EdgeID string `json:"edge_id"`
SourceNodeID string `json:"source_node_id"`
DesNodeId string `json:"des_node_id"`
SourcePortIndex int `json:"source_port_index"`
DesPortIndex int `json:"des_port_index"`
}
type MultiTypeValue struct {
Value any
}
// 实现json.Unmarshaler接口自定义解码逻辑
func (v *MultiTypeValue) UnmarshalJSON(data []byte) error {
// 尝试将数据解析为字符串
var strVal string
if err := json.Unmarshal(data, &strVal); err == nil {
v.Value = strVal
return nil
}
// 如果不是字符串,尝试解析为数字
var intVal int
if err := json.Unmarshal(data, &intVal); err == nil {
v.Value = intVal
return nil
}
// 如果不是字符串,尝试解析为数字
var boolVal bool
if err := json.Unmarshal(data, &boolVal); err == nil {
v.Value = boolVal
return nil
}
// 如果不是字符串,尝试解析为数字
var float64Val float64
if err := json.Unmarshal(data, &float64Val); err == nil {
v.Value = float64Val
return nil
}
var arrayVal []any
if err := json.Unmarshal(data, &arrayVal); err == nil {
v.Value = arrayVal
return nil
}
// 如果都失败,返回错误
return fmt.Errorf("cannot unmarshal JSON value: %s", string(data))
}
type variablesConfig struct {
Name string `json:"name"`
Type string `json:"type"`
Value MultiTypeValue `json:"value"`
}
type graphConfig struct {
GraphName string `json:"graph_name"`
Time string `json:"time"`
Nodes []nodeConfig `json:"nodes"`
Edges []edgeConfig `json:"edges"`
Variables []variablesConfig `json:"variables"`
}
func (gc *graphConfig) GetVariablesByName(varName string) *variablesConfig {
for _, varCfg := range gc.Variables {
if varCfg.Name == varName {
return &varCfg
}
}
return nil
}
func (gc *graphConfig) GetNodeByID(nodeID string) *nodeConfig {
for _, node := range gc.Nodes {
if node.Id == nodeID {
return &node
}
}
return nil
}
func (gr *Graph) Do(entranceID int64, args ...any) (Port_Array,error) {
entranceNode := gr.entrance[entranceID]
if entranceNode == nil {
return nil,fmt.Errorf("entranceID:%d not found", entranceID)
}
gr.variables = map[string]IPort{}
if gr.globalVariables == nil {
gr.globalVariables = map[string]IPort{}
}
err := entranceNode.Do(gr, args...)
if err != nil {
return nil, err
}
if gr.globalVariables!= nil {
port := gr.globalVariables[ReturnVarial]
if port != nil {
array,ok := port.GetArray()
if ok{
return array,nil
}
}
}
return nil,nil
}
func (gr *Graph) GetNodeInPortValue(nodeID string, inPortIndex int) IPort {
if ctx, ok := gr.context[nodeID]; ok {
if inPortIndex >= len(ctx.InputPorts) || inPortIndex < 0 {
return nil
}
return ctx.InputPorts[inPortIndex]
}
return nil
}
func (gr *Graph) GetNodeOutPortValue(nodeID string, outPortIndex int) IPort {
if ctx, ok := gr.context[nodeID]; ok {
if outPortIndex >= len(ctx.OutputPorts) || outPortIndex < 0 {
return nil
}
return ctx.OutputPorts[outPortIndex]
}
return nil
}
func (gr *Graph) Release() {
// 有定时器关闭定时器
for timerID := range gr.mapTimerID {
gr.CancelTimerId(gr.graphID, &timerID)
}
gr.mapTimerID = nil
// 清理掉所有数据
*gr = Graph{}
}

302
util/blueprint/graphpool.go Normal file
View File

@@ -0,0 +1,302 @@
package blueprint
import (
"fmt"
"github.com/goccy/go-json"
"os"
"path/filepath"
"strings"
)
type GraphPool struct {
mapGraphs map[string]*baseGraph
execPool *ExecPool
blueprintModule IBlueprintModule
}
func (gp *GraphPool) Load(execPool *ExecPool, graphFilePath string, blueprintModule IBlueprintModule) error {
gp.execPool = execPool
gp.mapGraphs = make(map[string]*baseGraph, 1024)
gp.blueprintModule = blueprintModule
// 检查路径是否存在
stat, err := os.Stat(graphFilePath)
if err != nil {
return fmt.Errorf("failed to access path %s: %v", graphFilePath, err)
}
// 如果是单个文件,直接处理
if !stat.IsDir() {
return fmt.Errorf("%s is not a directory", graphFilePath)
}
// 遍历目录及其子目录
return filepath.Walk(graphFilePath, func(path string, info os.FileInfo, err error) error {
if err != nil {
fmt.Printf("访问路径出错 %s: %v\n", path, err)
return nil // 继续遍历其他文件
}
// 如果是目录,继续遍历
if info.IsDir() {
return nil
}
// 只处理JSON文件
if filepath.Ext(path) == ".vgf" {
return gp.processJSONFile(path)
}
return nil
})
}
func (gp *GraphPool) Create(graphName string, graphID int64) IGraph {
gr, ok := gp.mapGraphs[graphName]
if !ok {
return nil
}
var graph Graph
graph.baseGraph = gr
graph.graphID = graphID
graph.context = make(map[string]*ExecContext, 4)
graph.IBlueprintModule = gp.blueprintModule
return &graph
}
func (gp *GraphPool) processJSONFile(filePath string) error {
// 打开文件
file, err := os.Open(filePath)
if err != nil {
return fmt.Errorf("failed to open file %s: %v", filePath, err)
}
defer func() {
if err := file.Close(); err != nil {
fmt.Printf("关闭文件 %s 时出错: %v\n", filePath, err)
}
}()
fileName := filepath.Base(filePath)
ext := filepath.Ext(fileName) // 获取".html"
name := strings.TrimSuffix(fileName, ext) // 获取"name"
var gConfig graphConfig
decoder := json.NewDecoder(file)
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)
}
func (gp *GraphPool) prepareGraph(graphName string, graphConfig *graphConfig) error {
// 找到所有的入口
for _, node := range graphConfig.Nodes {
_, entranceID, ok := getEntranceNodeNameAndID(node.Class)
if !ok {
continue
}
// 对入口进行预处理
err := gp.prepareOneEntrance(graphName, entranceID, &node, graphConfig)
if err != nil {
return err
}
}
return nil
}
func (gp *GraphPool) genVarExec(nodeCfg *nodeConfig, graphConfig *graphConfig) (IInnerExecNode, string) {
// 是否为Get_或Set_开头
if !strings.HasPrefix(nodeCfg.Class, "Get_") && !strings.HasPrefix(nodeCfg.Class, "Set_") {
return gp.execPool.GetExec(nodeCfg.Class), ""
}
// 获取Get_或Set_结尾字符串
var nodeName string
var varName string
if strings.HasPrefix(nodeCfg.Class, "Get_") {
var typ string
varName = strings.TrimPrefix(nodeCfg.Class, "Get_")
varCfg := graphConfig.GetVariablesByName(varName)
if varCfg != nil {
typ = varCfg.Type
}
nodeName = genGetVariablesNodeName(typ)
} else if strings.HasPrefix(nodeCfg.Class, "Set_") {
var typ string
varName = strings.TrimPrefix(nodeCfg.Class, "Set_")
varCfg := graphConfig.GetVariablesByName(varName)
if varCfg != nil {
typ = varCfg.Type
}
nodeName = genSetVariablesNodeName(typ)
}
e := gp.execPool.GetExec(nodeName)
e.(IExecNode).setVariableName(varName)
return e, varName
}
func (gp *GraphPool) genAllNode(graphConfig *graphConfig) (map[string]*execNode, error) {
nodes := make(map[string]*execNode)
for _, node := range graphConfig.Nodes {
var varName string
className := node.Class
if name, _, ok := getEntranceNodeNameAndID(className); ok {
className = name
}
// 获取不到node则获取变量node
exec := gp.execPool.GetExec(className)
if exec == nil {
exec, varName = gp.genVarExec(&node, graphConfig)
if exec == nil {
return nil, fmt.Errorf("%s node has not been registered", node.Class)
}
}
nodes[node.Id] = &execNode{
Id: node.Id,
execNode: exec,
preInPort: make([]*prePortNode, exec.GetInPortCount()),
inPortDefaultValue: node.PortDefault,
variableName: varName,
}
}
return nodes, nil
}
func (gp *GraphPool) prepareOneNode(mapNodeExec map[string]*execNode, nodeExec *execNode, graphConfig *graphConfig, recursion *int) error {
*recursion++
if *recursion > 100 {
return fmt.Errorf("recursion too deep")
}
// 找到所有出口
var idx int
for ; nodeExec.execNode.IsOutPortExec(idx) && idx < nodeExec.execNode.GetOutPortCount(); idx++ {
// 找到出口结点
nextExecNode := gp.findOutNextNode(graphConfig, mapNodeExec, nodeExec.Id, idx)
nodeExec.nextNode = append(nodeExec.nextNode, nextExecNode)
if nextExecNode != nil {
nextExecNode.beConnect = true
}
}
// 将所有的next填充next
for _, nextOne := range nodeExec.nextNode {
if nextOne == nil {
continue
}
// 对出口进行预处理
err := gp.prepareOneNode(mapNodeExec, nextOne, graphConfig, recursion)
if err != nil {
return err
}
}
return nil
}
func (gp *GraphPool) findOutNextNode(graphConfig *graphConfig, mapNodeExec map[string]*execNode, sourceNodeID string, sourcePortIdx int) *execNode {
// 找到出口的NodeID
for _, edge := range graphConfig.Edges {
if edge.SourceNodeID == sourceNodeID && edge.SourcePortIndex == sourcePortIdx {
return mapNodeExec[edge.DesNodeId]
}
}
return nil
}
// prepareOneEntrance 先处理执行Exec入出口连线
func (gp *GraphPool) prepareOneEntrance(graphName string, entranceID int64, nodeCfg *nodeConfig, graphConfig *graphConfig) error {
// 将所有的Node执行结点生成出来
mapNodes, err := gp.genAllNode(graphConfig)
if err != nil {
return err
}
// 从入口结点开始做预处理将next结点都统一生成
nodeExec := mapNodes[nodeCfg.Id]
if nodeExec == nil {
return fmt.Errorf("entrance node %s not found", nodeCfg.Id)
}
nodeExec.isEntrance = true
err = gp.prepareOneNode(mapNodes, nodeExec, graphConfig, new(int))
if err != nil {
return err
}
// 处理inPort前置结点
err = gp.prepareInPort(mapNodes, graphConfig)
if err != nil {
return err
}
gr, ok := gp.mapGraphs[graphName]
if !ok {
gr = &baseGraph{}
gr.entrance = make(map[int64]*execNode, 16)
gp.mapGraphs[graphName] = gr
}
gr.entrance[entranceID] = nodeExec
return nil
}
func (gp *GraphPool) findPreInPortNode(mapNodes map[string]*execNode, nodeExec *execNode, graphConfig *graphConfig, portIdx int) *prePortNode {
for _, edge := range graphConfig.Edges {
if edge.DesNodeId == nodeExec.Id && edge.DesPortIndex == portIdx {
srcNode := mapNodes[edge.SourceNodeID]
if srcNode == nil {
return nil
}
var preNode prePortNode
preNode.node = srcNode
preNode.outPortIndex = edge.SourcePortIndex
return &preNode
}
}
return nil
}
func (gp *GraphPool) preparePreInPortNode(mapNodes map[string]*execNode, nodeExec *execNode, graphConfig *graphConfig) error {
// 找到当前结点的所有inPort的前一个端口
for i := 0; i < nodeExec.execNode.GetInPortCount(); i++ {
// 如果是执行结点,则跳过
if nodeExec.execNode.IsInPortExec(i) {
continue
}
// 找到入口的上一个结点
preNode := gp.findPreInPortNode(mapNodes, nodeExec, graphConfig, i)
if preNode == nil {
continue
}
nodeExec.preInPort[i] = preNode
}
return nil
}
func (gp *GraphPool) prepareInPort(mapNodeExec map[string]*execNode, graphConfig *graphConfig) error {
for _, e := range mapNodeExec {
// 对当前结点的入口进行预处理
err := gp.preparePreInPortNode(mapNodeExec, e, graphConfig)
if err != nil {
return err
}
}
return nil
}

304
util/blueprint/mathnode.go Normal file
View File

@@ -0,0 +1,304 @@
package blueprint
import (
"fmt"
"math/rand"
)
func init() {
RegExecNode(&AddInt{})
RegExecNode(&SubInt{})
RegExecNode(&MulInt{})
RegExecNode(&DivInt{})
RegExecNode(&ModInt{})
RegExecNode(&RandNumber{})
}
// AddInt 加(int)
type AddInt struct {
BaseExecNode
}
func (em *AddInt) GetName() string {
return "AddInt"
}
func (em *AddInt) Exec() (int, error) {
inPortA := em.GetInPort(0)
if inPortA == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortB := em.GetInPort(1)
if inPortB == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inB, ok := inPortB.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
ret := inA + inB
outPortRet.SetInt(ret)
return -1, nil
}
// SubInt 减(int)
type SubInt struct {
BaseExecNode
}
func (em *SubInt) GetName() string {
return "SubInt"
}
func (em *SubInt) Exec() (int, error) {
inPortA := em.GetInPort(0)
if inPortA == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortB := em.GetInPort(1)
if inPortB == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
inPortAbs := em.GetInPort(2)
if inPortAbs == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inB, ok := inPortB.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
isAbs, ok := inPortAbs.GetBool()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
ret := inA - inB
if isAbs && ret < 0 {
ret *= -1
}
outPortRet.SetInt(ret)
return -1, nil
}
// MulInt 乘(int)
type MulInt struct {
BaseExecNode
}
func (em *MulInt) GetName() string {
return "MulInt"
}
func (em *MulInt) Exec() (int, error) {
inPortA := em.GetInPort(0)
if inPortA == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortB := em.GetInPort(1)
if inPortB == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inB, ok := inPortB.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet.SetInt(inA * inB)
return -1, nil
}
// DivInt 除(int)
type DivInt struct {
BaseExecNode
}
func (em *DivInt) GetName() string {
return "DivInt"
}
func (em *DivInt) Exec() (int, error) {
inPortA := em.GetInPort(0)
if inPortA == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortB := em.GetInPort(1)
if inPortB == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
inPortRound := em.GetInPort(2)
if inPortRound == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inB, ok := inPortB.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
isRound, ok := inPortRound.GetBool()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
if inB == 0 {
return -1, fmt.Errorf("div zero error")
}
var ret int64
if isRound {
ret = (inA + inB/2) / inB
} else {
ret = inA / inB
}
outPortRet.SetInt(ret)
return -1, nil
}
// ModInt 取模(int)
type ModInt struct {
BaseExecNode
}
func (em *ModInt) GetName() string {
return "ModInt"
}
func (em *ModInt) Exec() (int, error) {
inPortA := em.GetInPort(0)
if inPortA == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortB := em.GetInPort(1)
if inPortB == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inB, ok := inPortB.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
if inB == 0 {
return -1, fmt.Errorf("mod zero error")
}
outPortRet.SetInt(inA % inB)
return -1, nil
}
// RandNumber 范围随机[0,99]
type RandNumber struct {
BaseExecNode
}
func (em *RandNumber) GetName() string {
return "RandNumber"
}
func (em *RandNumber) Exec() (int, error) {
inPortSeed := em.GetInPort(0)
if inPortSeed == nil {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inPortMin := em.GetInPort(1)
if inPortMin == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
inPortMax := em.GetInPort(2)
if inPortMax == nil {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
outPortRet := em.GetOutPort(0)
if outPortRet == nil {
return -1, fmt.Errorf("AddInt outParam not found")
}
inSeed, ok := inPortSeed.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 1 not found")
}
inMin, ok := inPortMin.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
inMax, ok := inPortMax.GetInt()
if !ok {
return -1, fmt.Errorf("AddInt inParam 2 not found")
}
var ret int64
if inSeed > 0 {
r := rand.New(rand.NewSource(inSeed))
if r == nil {
return -1, fmt.Errorf("RandNumber fail")
}
ret = int64(r.Intn(int(inMax-inMin+1))) + inMin
} else {
ret = int64(rand.Intn(int(inMax-inMin+1))) + inMin
}
outPortRet.SetInt(ret)
return -1, nil
}

209
util/blueprint/node.go Normal file
View File

@@ -0,0 +1,209 @@
package blueprint
import (
"fmt"
)
type prePortNode struct {
node *execNode // 上个结点
outPortIndex int // 对应上一个结点的OutPort索引
}
type execNode struct {
Id string
execNode IInnerExecNode
nextNode []*execNode
nextIdx int
preInPort []*prePortNode // Port的上一个结点
inPortDefaultValue map[string]any
variableName string // 如果是变量,则有变量名
beConnect bool // 是否有被连线
isEntrance bool // 是否是入口结点
}
// HasBeConnectLine 是否有被连线
func (en *execNode) HasBeConnectLine() bool {
return en.beConnect
}
// HasInPortExec 有前置执行入口
func (en *execNode) HasInPortExec() bool {
return en.execNode.IsInPortExec(0)
}
// HasOutPortExec 有前置执行入口
func (en *execNode) HasOutPortExec() bool {
return en.execNode.IsOutPortExec(0)
}
func (en *execNode) GetInPortDefaultValue(index int) any {
key := fmt.Sprintf("%d", index)
v, ok := en.inPortDefaultValue[key]
if !ok {
return nil
}
return v
}
func (en *execNode) GetInPortDefaultIntArrayValue(index int) []int64 {
val := en.GetInPortDefaultValue(index)
if val == nil {
return nil
}
var arrayInt []int64
arrayVal := val.([]any)
for i := range arrayVal {
if intVal, ok := arrayVal[i].(float64); ok {
arrayInt = append(arrayInt, int64(intVal))
}
}
return arrayInt
}
func (en *execNode) GetInPortDefaultStringArrayValue(index int) []string {
val := en.GetInPortDefaultValue(index)
if val == nil {
return nil
}
return val.([]string)
}
func (en *execNode) Next() *execNode {
if en.nextIdx >= len(en.nextNode) {
return nil
}
return en.nextNode[en.nextIdx]
}
func (en *execNode) exec(gr *Graph) (int, error) {
e, ok := en.execNode.(IExecNode)
if !ok {
return -1, fmt.Errorf("exec node %s not exec", en.execNode.GetName())
}
node, ok := en.execNode.(IBaseExecNode)
if !ok {
return -1, fmt.Errorf("exec node %s not exec", en.execNode.GetName())
}
// 执行完要恢复上下文结点的BaseExecNode会被执行时修改
ctx, exNode := node.getExecNodeInfo()
defer func() {
node.setExecNodeInfo(ctx, exNode)
}()
if err := node.initExecNode(gr, en); err != nil {
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()
}
func (en *execNode) doSetInPort(gr *Graph, index int, inPort IPort) error {
// 找到当前Node的InPort的index的前一个结点
preNode := en.preInPort[index]
// 如果前一个结点为空,则填充默认值
if preNode == nil {
err := inPort.setAnyVale(en.GetInPortDefaultValue(index))
if err != nil {
return err
}
return nil
}
if _, ok := gr.context[preNode.node.Id]; !ok ||
(!preNode.node.HasBeConnectLine() && !preNode.node.isEntrance) {
// 如果前一个结点没有执行过,则递归执行前一个结点
err := preNode.node.Do(gr)
if err != nil {
return err
}
}
// 判断上一个结点是否已经执行过
if _, ok := gr.context[preNode.node.Id]; ok {
outPort := gr.GetNodeOutPortValue(preNode.node.Id, preNode.outPortIndex)
if outPort == nil {
return fmt.Errorf("pre node %s out port index %d not found", preNode.node.Id, preNode.outPortIndex)
}
inPort.SetValue(outPort)
return nil
}
return fmt.Errorf("pre node %s not exec", preNode.node.Id)
}
func (en *execNode) Do(gr *Graph, outPortArgs ...any) error {
// 重新初始化上下文
inPorts, outPorts := en.execNode.CloneInOutPort()
gr.context[en.Id] = &ExecContext{
InputPorts: inPorts,
OutputPorts: outPorts,
}
startOutIdx := en.execNode.GetOutPortParamStartIndex()
for i := 0; i < len(outPortArgs); i++ {
if i >= len(outPorts) {
return fmt.Errorf("args %d not found in node %s", i, en.execNode.GetName())
}
if err := outPorts[i+startOutIdx].setAnyVale(outPortArgs[i]); err != nil {
return fmt.Errorf("args %d set value error: %w", i, err)
}
}
// 处理InPort结点值
var err error
for index := range inPorts {
if en.execNode.IsInPortExec(index) {
continue
}
err = en.doSetInPort(gr, index, inPorts[index])
if err != nil {
return err
}
}
// 设置执行器相关的上下文信息
// 如果是变量设置变量名
// 执行本结点
nextIndex, err := en.exec(gr)
if err != nil {
return err
}
if nextIndex == -1 || en.nextNode == nil {
return nil
}
if nextIndex < 0 || nextIndex >= len(en.nextNode) {
return fmt.Errorf("next index %d not found", nextIndex)
}
if en.nextNode[nextIndex] == nil {
return nil
}
return en.nextNode[nextIndex].Do(gr)
}

420
util/blueprint/port.go Normal file
View File

@@ -0,0 +1,420 @@
package blueprint
import (
"fmt"
"strconv"
)
const (
Config_PortType_Exec = "exec"
Config_PortType_Data = "data"
Config_DataType_Int = "int"
Config_DataType_Integer = "integer"
Config_DataType_Float = "float"
Config_DataType_Str = "string"
Config_DataType_Boolean = "boolean"
Config_DataType_Bool = "bool"
Config_DataType_Array = "array"
)
type Port[T iPortType] struct {
PortVal T
}
func (em *Port[T]) Clone() IPort {
arrayData, ok := any(em.PortVal).(Port_Array)
if !ok {
return &Port[T]{
PortVal: em.PortVal,
}
}
portArray := Port[Port_Array]{}
portArray.PortVal = append(portArray.PortVal, arrayData...)
return &portArray
}
func (em *Port[T]) Reset() {
var v T
em.PortVal = v
}
func (em *Port[T]) GetInt() (Port_Int, bool) {
if t, ok := any(em.PortVal).(Port_Int); ok {
return t, true
}
return 0, false
}
func (em *Port[T]) GetFloat() (Port_Float, bool) {
if t, ok := any(em.PortVal).(Port_Float); ok {
return t, true
}
return 0, false
}
func (em *Port[T]) GetStr() (Port_Str, bool) {
if t, ok := any(em.PortVal).(Port_Str); ok {
return t, true
}
return "", false
}
func (em *Port[T]) GetArrayValInt(idx int) (Port_Int, bool) {
if t, ok := any(em.PortVal).(Port_Array); ok {
if idx >= 0 && idx < len(t) {
return t[idx].IntVal, true
}
}
return 0, false
}
func (em *Port[T]) GetArrayValStr(idx int) (string, bool) {
if t, ok := any(em.PortVal).(Port_Array); ok {
if idx >= 0 && idx < len(t) {
return t[idx].StrVal, true
}
}
return "", false
}
func (em *Port[T]) GetBool() (Port_Bool, bool) {
if t, ok := any(em.PortVal).(Port_Bool); ok {
return t, true
}
return false, false
}
func (em *Port[T]) GetArray() (Port_Array, bool) {
if t, ok := any(em.PortVal).(Port_Array); ok {
return t, true
}
return nil, false
}
func (em *Port[T]) SetInt(val Port_Int) bool {
if t, ok := any(&em.PortVal).(*Port_Int); ok {
*t = val
return true
}
return false
}
func (em *Port[T]) SetFloat(val Port_Float) bool {
if t, ok := any(&em.PortVal).(*Port_Float); ok {
*t = val
return true
}
return false
}
func (em *Port[T]) SetStr(val Port_Str) bool {
if t, ok := any(&em.PortVal).(*Port_Str); ok {
*t = val
return true
}
return false
}
func (em *Port[T]) SetBool(val Port_Bool) bool {
if t, ok := any(&em.PortVal).(*Port_Bool); ok {
*t = val
return true
}
return false
}
func (em *Port[T]) SetArrayValInt(idx int, val Port_Int) bool {
if t, ok := any(em.PortVal).(Port_Array); ok {
if idx >= 0 && idx < len(t) {
t[idx].IntVal = val
return true
}
}
return false
}
func (em *Port[T]) SetArrayValStr(idx int, val Port_Str) bool {
if t, ok := any(em.PortVal).(Port_Array); ok {
if idx >= 0 && idx < len(t) {
(t)[idx].StrVal = val
return true
}
}
return false
}
func (em *Port[T]) AppendArrayValInt(val Port_Int) bool {
if t, ok := any(&em.PortVal).(*Port_Array); ok {
*t = append(*t, ArrayData{IntVal: val})
return true
}
return false
}
func (em *Port[T]) AppendArrayValStr(val Port_Str) bool {
if t, ok := any(&em.PortVal).(*Port_Array); ok {
*t = append(*t, ArrayData{StrVal: val})
return true
}
return false
}
func (em *Port[T]) AppendArrayData(val ArrayData) bool {
if t, ok := any(&em.PortVal).(*Port_Array); ok {
*t = append(*t, val)
return true
}
return false
}
func (em *Port[T]) GetArrayLen() Port_Int {
if t, ok := any(&em.PortVal).(*Port_Array); ok {
return Port_Int(len(*t))
}
return 0
}
func (em *Port[T]) IsPortExec() bool {
_, ok := any(em.PortVal).(Port_Exec)
return ok
}
func (em *Port[T]) convertInt64(v any) (int64, bool) {
switch v.(type) {
case int:
return int64(v.(int)), true
case int64:
return v.(int64), true
case int32:
return int64(v.(int32)), true
case int16:
return int64(v.(int16)), true
case int8:
return int64(v.(int8)), true
case uint64:
return int64(v.(uint64)), true
case uint32:
return int64(v.(uint32)), true
case uint16:
return int64(v.(uint16)), true
case uint8:
return int64(v.(uint8)), true
case uint:
return int64(v.(uint)), true
default:
return 0, false
}
}
func (em *Port[T]) setAnyVale(v any) error {
switch v.(type) {
case int, int64:
val, ok := em.convertInt64(v)
if !ok {
return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)
}
switch any(em.PortVal).(type) {
case Port_Int:
em.SetInt(val)
case Port_Float:
em.SetFloat(Port_Float(val))
case Port_Str:
em.SetStr(fmt.Sprintf("%d", int64(val)))
case Port_Bool:
em.SetBool(int64(val) != 0)
default:
return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)
}
case float64:
fV := v.(float64)
switch any(em.PortVal).(type) {
case Port_Int:
em.SetInt(Port_Int(fV))
case Port_Float:
em.SetFloat(fV)
case Port_Str:
em.SetStr(fmt.Sprintf("%d", int64(fV)))
case Port_Bool:
em.SetBool(int64(fV) != 0)
default:
return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)
}
case string:
strV := v.(string)
switch any(em.PortVal).(type) {
case Port_Int:
val, err := strconv.Atoi(strV)
if err != nil {
return err
}
em.SetInt(Port_Int(val))
case Port_Float:
fV, err := strconv.ParseFloat(strV, 64)
if err != nil {
return err
}
em.SetFloat(fV)
case Port_Str:
em.SetStr(strV)
case Port_Bool:
val, err := strconv.ParseBool(strV)
if err != nil {
return err
}
em.SetBool(val)
default:
return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)
}
case bool:
strV := v.(bool)
switch any(em.PortVal).(type) {
case Port_Int:
return fmt.Errorf("port type is int, but value is %v", strV)
case Port_Float:
return fmt.Errorf("port type is float, but value is %v", strV)
case Port_Str:
return fmt.Errorf("port type is string, but value is %v", strV)
case Port_Bool:
em.SetBool(strV)
default:
return fmt.Errorf("port type is %T, but value is %v", em.PortVal, v)
}
case []int64:
arr := v.([]int64)
for _, val := range arr {
em.AppendArrayValInt(val)
}
case []int32:
arr := v.([]int32)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []int16:
arr := v.([]int16)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []int8:
arr := v.([]int8)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []uint64:
arr := v.([]uint64)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []uint32:
arr := v.([]uint32)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []uint16:
arr := v.([]uint16)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []uint8:
arr := v.([]uint8)
for _, val := range arr {
em.AppendArrayValInt(Port_Int(val))
}
case []string:
arr := v.([]string)
for _, val := range arr {
em.AppendArrayValStr(val)
}
case Port_Array:
arr := v.(Port_Array)
for _, val := range arr {
em.AppendArrayValInt(val.IntVal)
}
}
return nil
}
func (em *Port[T]) SetValue(val IPort) bool {
valT, ok := val.(*Port[T])
if !ok {
return false
}
em.PortVal = valT.PortVal
return true
}
type IPort interface {
GetInt() (Port_Int, bool)
GetFloat() (Port_Float, bool)
GetStr() (Port_Str, bool)
GetArrayValInt(idx int) (Port_Int, bool)
GetArrayValStr(idx int) (Port_Str, bool)
GetBool() (Port_Bool, bool)
GetArray() (Port_Array, bool)
SetInt(val Port_Int) bool
SetFloat(val Port_Float) bool
SetStr(val Port_Str) bool
SetBool(val Port_Bool) bool
SetArrayValInt(idx int, val Port_Int) bool
SetArrayValStr(idx int, val Port_Str) bool
AppendArrayValInt(val Port_Int) bool
AppendArrayValStr(val Port_Str) bool
GetArrayLen() Port_Int
Clone() IPort
Reset()
IsPortExec() bool
setAnyVale(v any) error
SetValue(val IPort) bool
}
func NewPortExec() IPort {
return &Port[Port_Exec]{}
}
func NewPortInt() IPort {
return &Port[Port_Int]{}
}
func NewPortFloat() IPort {
return &Port[Port_Float]{}
}
func NewPortStr() IPort {
return &Port[Port_Str]{}
}
func NewPortBool() IPort {
return &Port[Port_Bool]{}
}
func NewPortArray() IPort {
return &Port[Port_Array]{}
}
func NewPortByType(typ string) IPort {
switch typ {
case Config_PortType_Exec:
return NewPortExec()
case Config_DataType_Int, Config_DataType_Integer:
return NewPortInt()
case Config_DataType_Float:
return NewPortFloat()
case Config_DataType_Str:
return NewPortStr()
case Config_DataType_Bool, Config_DataType_Boolean:
return NewPortBool()
case Config_DataType_Array:
return NewPortArray()
default:
return nil
}
}

View File

@@ -0,0 +1,7 @@
package blueprint
var execNodes []IExecNode
func RegExecNode(exec IExecNode) {
execNodes = append(execNodes, exec)
}

689
util/blueprint/sysnodes.go Normal file
View File

@@ -0,0 +1,689 @@
package blueprint
import (
"fmt"
"github.com/duanhf2012/origin/v2/log"
"math/rand/v2"
"time"
)
// 系统入口ID定义1000以内
const (
EntranceID_IntParam = 1
EntranceID_ArrayParam = 2
EntranceID_Timer = 3
)
func init() {
RegExecNode(&Entrance_ArrayParam{})
RegExecNode(&Entrance_IntParam{})
RegExecNode(&Entrance_Timer{})
RegExecNode(&Output{})
RegExecNode(&Sequence{})
RegExecNode(&Foreach{})
RegExecNode(&GetArrayInt{})
RegExecNode(&GetArrayString{})
RegExecNode(&GetArrayLen{})
RegExecNode(&CreateIntArray{})
RegExecNode(&CreateStringArray{})
RegExecNode(&AppendIntegerToArray{})
RegExecNode(&AppendStringToArray{})
RegExecNode(&BoolIf{})
RegExecNode(&GreaterThanInteger{})
RegExecNode(&LessThanInteger{})
RegExecNode(&EqualInteger{})
RegExecNode(&RangeCompare{})
RegExecNode(&Probability{})
RegExecNode(&CreateTimer{})
}
type Entrance_ArrayParam struct {
BaseExecNode
}
func (em *Entrance_ArrayParam) GetName() string {
return "Entrance_ArrayParam"
}
func (em *Entrance_ArrayParam) Exec() (int, error) {
return 0, nil
}
type Entrance_IntParam struct {
BaseExecNode
}
func (em *Entrance_IntParam) GetName() string {
return "Entrance_IntParam"
}
func (em *Entrance_IntParam) Exec() (int, error) {
return 0, nil
}
type Entrance_Timer struct {
BaseExecNode
}
func (em *Entrance_Timer) GetName() string {
return "Entrance_Timer"
}
func (em *Entrance_Timer) Exec() (int, error) {
return 0, nil
}
type Output struct {
BaseExecNode
}
func (em *Output) GetName() string {
return "Output"
}
func (em *Output) Exec() (int, error) {
val, ok := em.GetInPortInt(1)
if !ok {
return 0, fmt.Errorf("output Exec inParam not found")
}
valStr, ok := em.GetInPortStr(2)
if !ok {
return 0, fmt.Errorf("output Exec inParam not found")
}
valArray, ok := em.GetInPortArray(3)
if !ok {
return 0, fmt.Errorf("output Exec inParam not found")
}
fmt.Printf("output Exec inParam [%d] [%s] [%v]\n", val, valStr, valArray)
return 0, nil
}
type Sequence struct {
BaseExecNode
}
func (em *Sequence) GetName() string {
return "Sequence"
}
func (em *Sequence) Exec() (int, error) {
for i := range em.outPort {
if !em.outPort[i].IsPortExec() {
break
}
err := em.DoNext(i)
if err != nil {
return -1, err
}
}
return -1, nil
}
type Foreach struct {
BaseExecNode
}
func (em *Foreach) GetName() string {
return "Foreach"
}
func (em *Foreach) Exec() (int, error) {
startIndex, ok := em.ExecContext.InputPorts[1].GetInt()
if !ok {
return 0, fmt.Errorf("foreach Exec inParam not found")
}
endIndex, ok := em.ExecContext.InputPorts[2].GetInt()
if !ok {
return 0, fmt.Errorf("foreach Exec inParam not found")
}
for i := startIndex; i < endIndex; i++ {
em.ExecContext.OutputPorts[2].SetInt(i)
err := em.DoNext(0)
if err != nil {
return -1, err
}
}
err := em.DoNext(1)
if err != nil {
return -1, err
}
return -1, nil
}
type GetArrayInt struct {
BaseExecNode
}
func (em *GetArrayInt) GetName() string {
return "GetArrayInt"
}
func (em *GetArrayInt) Exec() (int, error) {
inPort := em.GetInPort(0)
if inPort == nil {
return -1, fmt.Errorf("GetArrayInt inParam not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("GetArrayInt outParam not found")
}
arrIndexPort := em.GetInPort(1)
if arrIndexPort == nil {
return -1, fmt.Errorf("GetArrayInt arrIndexParam not found")
}
arrIndex, ok := arrIndexPort.GetInt()
if !ok {
return -1, fmt.Errorf("GetArrayInt arrIndexParam not found")
}
if arrIndex < 0 || arrIndex >= inPort.GetArrayLen() {
return -1, fmt.Errorf("GetArrayInt arrIndexParam out of range,index %d", arrIndex)
}
val, ok := inPort.GetArrayValInt(int(arrIndex))
if !ok {
log.Errorf("GetArrayValInt failed, idx:%d", arrIndex)
return -1, fmt.Errorf("GetArrayInt inParam not found")
}
outPort.SetInt(val)
return -1, nil
}
type GetArrayString struct {
BaseExecNode
}
func (em *GetArrayString) GetName() string {
return "GetArrayString"
}
func (em *GetArrayString) Exec() (int, error) {
inPort := em.GetInPort(0)
if inPort == nil {
return -1, fmt.Errorf("GetArrayInt inParam 0 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("GetArrayInt outParam 0 not found")
}
arrIndexPort := em.GetInPort(1)
if arrIndexPort == nil {
return -1, fmt.Errorf("GetArrayInt arrIndexParam 1 not found")
}
arrIndex, ok := arrIndexPort.GetInt()
if !ok {
return -1, fmt.Errorf("GetArrayInt arrIndexParam not found")
}
if arrIndex < 0 || arrIndex >= inPort.GetArrayLen() {
return -1, fmt.Errorf("GetArrayInt arrIndexParam out of range,index %d", arrIndex)
}
val, ok := inPort.GetArrayValStr(int(arrIndex))
if !ok {
log.Errorf("GetArrayValStr failed, idx:%d", arrIndex)
return -1, fmt.Errorf("GetArrayInt inParam not found")
}
outPort.SetStr(val)
return -1, nil
}
type GetArrayLen struct {
BaseExecNode
}
func (em *GetArrayLen) GetName() string {
return "GetArrayLen"
}
func (em *GetArrayLen) Exec() (int, error) {
inPort := em.GetInPort(0)
if inPort == nil {
return -1, fmt.Errorf("GetArrayInt inParam 0 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("GetArrayInt outParam 0 not found")
}
outPort.SetInt(inPort.GetArrayLen())
return -1, nil
}
// BoolIf 布尔判断
type BoolIf struct {
BaseExecNode
}
func (em *BoolIf) GetName() string {
return "BoolIf"
}
func (em *BoolIf) Exec() (int, error) {
inPort := em.GetInPort(1)
if inPort == nil {
return -1, fmt.Errorf("GetArrayInt inParam 1 not found")
}
ret, ok := inPort.GetBool()
if !ok {
return -1, fmt.Errorf("BoolIf inParam error")
}
if ret {
return 1, nil
}
return 0, nil
}
// GreaterThanInteger 大于(整型) >
type GreaterThanInteger struct {
BaseExecNode
}
func (em *GreaterThanInteger) GetName() string {
return "GreaterThanInteger"
}
func (em *GreaterThanInteger) Exec() (int, error) {
inPortEqual := em.GetInPort(1)
if inPortEqual == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inPortA := em.GetInPort(2)
if inPortA == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inPorB := em.GetInPort(3)
if inPorB == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
ret, ok := inPortEqual.GetBool()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 error")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 2 error")
}
inB, ok := inPorB.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 3 error")
}
if ret {
if inA >= inB {
return 1, nil
}
return 0, nil
}
if inA > inB {
return 1, nil
}
return 0, nil
}
// LessThanInteger 小于(整型) <
type LessThanInteger struct {
BaseExecNode
}
func (em *LessThanInteger) GetName() string {
return "LessThanInteger"
}
func (em *LessThanInteger) Exec() (int, error) {
inPortEqual := em.GetInPort(1)
if inPortEqual == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inPortA := em.GetInPort(2)
if inPortA == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inPorB := em.GetInPort(3)
if inPorB == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
ret, ok := inPortEqual.GetBool()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 error")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 2 error")
}
inB, ok := inPorB.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 3 error")
}
if ret {
if inA <= inB {
return 1, nil
}
return 0, nil
}
if inA < inB {
return 1, nil
}
return 0, nil
}
// EqualInteger 等于(整型)==
type EqualInteger struct {
BaseExecNode
}
func (em *EqualInteger) GetName() string {
return "EqualInteger"
}
func (em *EqualInteger) Exec() (int, error) {
inPortA := em.GetInPort(1)
if inPortA == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inPorB := em.GetInPort(2)
if inPorB == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inA, ok := inPortA.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 2 error")
}
inB, ok := inPorB.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 3 error")
}
if inA == inB {
return 1, nil
}
return 0, nil
}
// RangeCompare 范围比较<=
type RangeCompare struct {
BaseExecNode
}
func (em *RangeCompare) GetName() string {
return "RangeCompare"
}
func (em *RangeCompare) 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
}
func (em *Probability) GetName() string {
return "Probability"
}
func (em *Probability) Exec() (int, error) {
inPortProbability := em.GetInPort(1)
if inPortProbability == nil {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 not found")
}
inProbability, ok := inPortProbability.GetInt()
if !ok {
return -1, fmt.Errorf("GreaterThanInteger inParam 1 error")
}
if inProbability > rand.Int64N(10000) {
return 1, nil
}
return 0, nil
}
// CreateIntArray 创建整型数组
type CreateIntArray struct {
BaseExecNode
}
func (em *CreateIntArray) GetName() string {
return "CreateIntArray"
}
func (em *CreateIntArray) Exec() (int, error) {
intArray := em.execNode.GetInPortDefaultIntArrayValue(0)
if intArray == nil {
return -1, fmt.Errorf("CreateIntArray inParam 0 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("GetArrayInt outParam 0 not found")
}
for _, v := range intArray {
outPort.AppendArrayValInt(v)
}
return -1, nil
}
// CreateStringArray 创建字符串数组
type CreateStringArray struct {
BaseExecNode
}
func (em *CreateStringArray) GetName() string {
return "CreateStringArray"
}
func (em *CreateStringArray) Exec() (int, error) {
intArray := em.execNode.GetInPortDefaultStringArrayValue(0)
if intArray == nil {
return -1, fmt.Errorf("CreateIntArray inParam 0 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("GetArrayInt outParam 0 not found")
}
for _, v := range intArray {
outPort.AppendArrayValStr(v)
}
return -1, nil
}
// AppendIntegerToArray 数组追加整型
type AppendIntegerToArray struct {
BaseExecNode
}
func (em *AppendIntegerToArray) GetName() string {
return "AppendIntegerToArray"
}
func (em *AppendIntegerToArray) Exec() (int, error) {
inPortArray := em.GetInPort(0)
if inPortArray == nil {
return -1, fmt.Errorf("AppendIntegerToArray inParam 0 not found")
}
inPortVal := em.GetInPort(1)
if inPortVal == nil {
return -1, fmt.Errorf("AppendIntegerToArray inParam 1 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("AppendIntegerToArray outParam 0 not found")
}
intArray, ok := inPortArray.GetArray()
if !ok {
return -1, fmt.Errorf("AppendIntegerToArray inParam 0 error")
}
intVal, ok := inPortVal.GetInt()
if !ok {
return -1, fmt.Errorf("AppendIntegerToArray inParam 1 error")
}
for i := range intArray {
outPort.AppendArrayValInt(intArray[i].IntVal)
}
outPort.AppendArrayValInt(intVal)
return -1, nil
}
// AppendStringToArray 数组追加字符串
type AppendStringToArray struct {
BaseExecNode
}
func (em *AppendStringToArray) GetName() string {
return "AppendStringToArray"
}
func (em *AppendStringToArray) Exec() (int, error) {
inPortArray := em.GetInPort(0)
if inPortArray == nil {
return -1, fmt.Errorf("AppendStringToArray inParam 0 not found")
}
inPortVal := em.GetInPort(1)
if inPortVal == nil {
return -1, fmt.Errorf("AppendStringToArray inParam 1 not found")
}
outPort := em.GetOutPort(0)
if outPort == nil {
return -1, fmt.Errorf("AppendStringToArray outParam 0 not found")
}
intArray, ok := inPortArray.GetArray()
if !ok {
return -1, fmt.Errorf("AppendStringToArray inParam 0 error")
}
for i := range intArray {
outPort.AppendArrayValStr(intArray[i].StrVal)
}
return -1, nil
}
// CreateTimer 创建定时器
type CreateTimer struct {
BaseExecNode
}
func (em *CreateTimer) GetName() string {
return "CreateTimer"
}
func (em *CreateTimer) Exec() (int, error) {
delay, ok := em.GetInPortInt(0)
if !ok {
return -1, fmt.Errorf("CreateTimer inParam 0 error")
}
array, ok := em.GetInPortArray(1)
if !ok {
return -1, fmt.Errorf("CreateTimer inParam 0 error")
}
var timerId uint64
graphID := em.gr.graphID
em.gr.IBlueprintModule.SafeAfterFunc(&timerId, time.Duration(delay)*time.Millisecond, nil, func(timerId uint64, additionData interface{}) {
err := em.gr.IBlueprintModule.TriggerEvent(graphID, EntranceID_Timer, array)
if err != nil {
log.Warnf("CreateTimer SafeAfterFunc error timerId:%d err:%v", timerId, err)
}
em.gr.IBlueprintModule.CancelTimerId(graphID,&timerId)
})
em.gr.mapTimerID[timerId] = struct{}{}
outPort := em.GetOutPort(1)
if outPort == nil {
return -1, fmt.Errorf("CreateTimer outParam 1 not found")
}
outPort.SetInt(int64(timerId))
return 0, nil
}
// CloseTimer 关闭定时器
type CloseTimer struct {
BaseExecNode
}
func (em *CloseTimer) GetName() string {
return "CloseTimer"
}
func (em *CloseTimer) Exec() (int, error) {
timerID, ok := em.GetInPortInt(1)
if !ok {
return -1, fmt.Errorf("CreateTimer inParam 0 error")
}
id := uint64(timerID)
ok = em.gr.IBlueprintModule.CancelTimerId(em.gr.graphID, &id)
if !ok {
log.Warnf("CloseTimer CancelTimerId:%d", id)
}
return 0, nil
}

25
util/blueprint/typedef.go Normal file
View File

@@ -0,0 +1,25 @@
package blueprint
type ArrayElement struct {
IntVal int64
StrVal string
}
type PortExec struct {
}
type ArrayData struct {
IntVal int64
StrVal string
}
type Port_Exec = PortExec
type Port_Int = int64
type Port_Float = float64
type Port_Str = string
type Port_Bool = bool
type Port_Array []ArrayData
type iPortType interface {
Port_Exec | Port_Int | Port_Float | Port_Str | Port_Bool | Port_Array
}

View File

@@ -0,0 +1,79 @@
package blueprint
import (
"fmt"
"strings"
)
const GetVariables = "GetVar"
const SetVariables = "SetVar"
const globalVariablesPrefix = "g_"
type GetVariablesNode struct {
BaseExecNode
nodeName string
varName string
}
type SetVariablesNode struct {
BaseExecNode
nodeName string
varName string
}
func (g *GetVariablesNode) GetName() string {
return g.nodeName
}
func (g *GetVariablesNode) Exec() (int, error) {
var port IPort
if strings.HasPrefix(g.varName, globalVariablesPrefix) {
port = g.gr.globalVariables[g.varName]
} else {
port = g.gr.variables[g.varName]
}
if port == nil {
return -1, fmt.Errorf("variable %s not found,node name %s", g.varName, g.nodeName)
}
if !g.SetOutPort(0, port) {
return -1, fmt.Errorf("set out port failed,node name %s", g.nodeName)
}
return 0, nil
}
func (g *GetVariablesNode) setVariableName(name string) bool {
g.varName = name
return true
}
func (g *SetVariablesNode) GetName() string {
return g.nodeName
}
func (g *SetVariablesNode) Exec() (int, error) {
port := g.GetInPort(1)
if port == nil {
return -1, fmt.Errorf("get in port failed,node name %s", g.nodeName)
}
varPort := port.Clone()
if strings.HasPrefix(g.varName, globalVariablesPrefix) {
g.gr.globalVariables[g.varName] = varPort
} else {
g.gr.variables[g.varName] = varPort
}
if !g.SetOutPort(1, varPort) {
return -1, fmt.Errorf("set out port failed,node name %s", g.nodeName)
}
return 0, nil
}
func (g *SetVariablesNode) setVariableName(name string) bool {
g.varName = name
return true
}

View File

@@ -32,10 +32,10 @@ func Abs[NumType typ.Signed | typ.Float](Num NumType) NumType {
func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
ret := number1 + number2
if number2 > 0 && ret < number1 {
log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
return ret, false
} else if number2 < 0 && ret > number1 {
log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
return ret, false
}
@@ -45,10 +45,10 @@ func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
func SubSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
ret := number1 - number2
if number2 > 0 && ret > number1 {
log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
return ret, false
} else if number2 < 0 && ret < number1 {
log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
return ret, false
}
@@ -65,7 +65,7 @@ func MulSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
return ret, true
}
log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
return ret, true
}

41
util/stack/stack.go Normal file
View File

@@ -0,0 +1,41 @@
package stack
// Stack 是一个通用类型的栈结构
type Stack[T any] struct {
items []*T
}
// Push 将元素压入栈顶
func (s *Stack[T]) Push(item *T) {
s.items = append(s.items, item)
}
// Pop 弹出并返回栈顶元素,如果栈为空则返回错误
func (s *Stack[T]) Pop() *T {
if len(s.items) == 0 {
return nil
}
index := len(s.items) - 1
item := s.items[index]
s.items = s.items[:index]
return item
}
// Peek 返回栈顶元素但不移除,如果栈为空则返回错误
func (s *Stack[T]) Peek() *T {
if len(s.items) == 0 {
return nil
}
return s.items[len(s.items)-1]
}
// IsEmpty 检查栈是否为空
func (s *Stack[T]) IsEmpty() bool {
return len(s.items) == 0
}
// Size 返回栈中元素的数量
func (s *Stack[T]) Size() int {
return len(s.items)
}

View File

@@ -6,21 +6,21 @@ import (
"time"
)
func SetupTimer(timer ITimer) ITimer{
func SetupTimer(timer ITimer) ITimer {
if timer.IsOpen() == true {
return nil
}
timer.Open(true)
timerHeapLock.Lock() // 使用锁规避竞争条件
heap.Push(&timerHeap,timer)
heap.Push(&timerHeap, timer)
timerHeapLock.Unlock()
return timer
}
func NewTimer(d time.Duration) *Timer{
c := make(chan ITimer,1)
timer := newTimer(d,c,nil,"")
func NewTimer(d time.Duration) *Timer {
c := make(chan ITimer, 1)
timer := newTimer(d, c, nil, "")
SetupTimer(timer)
return timer
@@ -43,7 +43,7 @@ func (h *_TimerHeap) Less(i, j int) bool {
}
func (h *_TimerHeap) Swap(i, j int) {
h.timers[i],h.timers[j] = h.timers[j],h.timers[i]
h.timers[i], h.timers[j] = h.timers[j], h.timers[i]
}
func (h *_TimerHeap) Push(x interface{}) {
@@ -62,16 +62,16 @@ var (
timeOffset time.Duration
)
func StartTimer(minTimerInterval time.Duration,maxTimerNum int){
timerHeap.timers = make([]ITimer,0,maxTimerNum)
func StartTimer(minTimerInterval time.Duration, maxTimerNum int) {
timerHeap.timers = make([]ITimer, 0, maxTimerNum)
heap.Init(&timerHeap) // 初始化定时器heap
go tickRoutine(minTimerInterval)
go tickRoutine(minTimerInterval)
}
func tickRoutine(minTimerInterval time.Duration){
func tickRoutine(minTimerInterval time.Duration) {
var bContinue bool
for{
for {
bContinue = tick()
if bContinue == false {
time.Sleep(minTimerInterval)
@@ -79,7 +79,7 @@ func tickRoutine(minTimerInterval time.Duration){
}
}
func tick() bool{
func tick() bool {
now := Now()
timerHeapLock.Lock()
if timerHeap.Len() <= 0 { // 没有任何定时器,立刻返回
@@ -100,7 +100,7 @@ func tick() bool{
return true
}
func Now() time.Time{
func Now() time.Time {
if timeOffset == 0 {
return time.Now()
}
@@ -108,4 +108,12 @@ func Now() time.Time{
return time.Now().Add(timeOffset)
}
// SetTimeOffset 设置时间偏移量
func SetTimeOffset(offset time.Duration) {
timeOffset = offset
}
// AddTimeOffset 添加时间偏移量
func AddTimeOffset(addOffset time.Duration) {
timeOffset += addOffset
}