[feat]添加规则引擎debug功能

apps/device/services/device_alarm.go

@@ -11,7 +11,7 @@ type (
 	DeviceAlarmModel interface {
 		Insert(data entity.DeviceAlarm) error
 		FindOne(id string) *entity.DeviceAlarm
-		FindOneByType(deviceId, ty, state string) *entity.DeviceAlarm
+		FindOneByType(deviceId, ty, state string) (*entity.DeviceAlarm, error)
 		FindListPage(page, pageSize int, data entity.DeviceAlarmForm) (*[]entity.DeviceAlarm, int64)
 		Update(data entity.DeviceAlarm) error
 		Delete(ids []string)
@@ -41,12 +41,14 @@ func (m *alarmModelImpl) FindOne(id string) *entity.DeviceAlarm {
 	return resData
 }
 
-func (m *alarmModelImpl) FindOneByType(deviceId, ty, state string) *entity.DeviceAlarm {
+func (m *alarmModelImpl) FindOneByType(deviceId, ty, state string) (*entity.DeviceAlarm, error) {
 	resData := new(entity.DeviceAlarm)
 	db := global.Db.Table(m.table).Where("device_id = ?", deviceId).Where("type = ? ", ty).Where("state = ? ", state)
 	err := db.First(resData).Error
-	biz.ErrIsNil(err, "查询设备告警失败")
-	return resData
+	if err != nil {
+		return nil, err
+	}
+	return resData, nil
 }
 
 func (m *alarmModelImpl) FindListPage(page, pageSize int, data entity.DeviceAlarmForm) (*[]entity.DeviceAlarm, int64) {

apps/device/util/device_rpc.go

@@ -28,7 +28,7 @@ func BuildRunDeviceRpc(deviceId, mode string, metadata map[string]interface{}) e
 	dataCode := ruleData.LfData.DataCode
 	code, _ := json.Marshal(dataCode)
 	//新建规则链实体
-	instance, errs := rule_engine.NewRuleChainInstance(code)
+	instance, errs := rule_engine.NewRuleChainInstance(findOne.Id, code)
 	if len(errs) > 0 {
 		return errs[0]
 	}

apps/rule/api/rulechain.go

@@ -1,7 +1,6 @@
 package api
 
 import (
-	"context"
 	"github.com/PandaXGO/PandaKit/biz"
 	"github.com/PandaXGO/PandaKit/model"
 	"github.com/PandaXGO/PandaKit/restfulx"
@@ -9,8 +8,6 @@ import (
 	"pandax/apps/rule/services"
 	"pandax/pkg/global_model"
 	"pandax/pkg/rule_engine"
-	"pandax/pkg/rule_engine/message"
-	"pandax/pkg/rule_engine/nodes"
 	"strings"
 )
 
@@ -19,14 +16,22 @@ type RuleChainApi struct {
 }
 
 func (r *RuleChainApi) GetNodeLabels(rc *restfulx.ReqCtx) {
-	rc.ResData = nodes.GetCategory()
+	rc.ResData = rule_engine.GetCategory()
 }
-func (r *RuleChainApi) RuleChainTest(rc *restfulx.ReqCtx) {
-	code := restfulx.QueryParam(rc, "code")
-	instance, _ := rule_engine.NewRuleChainInstance([]byte(code))
-	msg := message.NewMessage("1", message.TelemetryMes, message.Msg{"temperature": 60.4, "humidity": 32.5}, message.Metadata{})
-	instance.StartRuleChain(context.Background(), msg)
-	rc.ResData = []map[string]interface{}{}
+
+func (r *RuleChainApi) GetNodeDebug(rc *restfulx.ReqCtx) {
+	pageNum := restfulx.QueryInt(rc, "pageNum", 1)
+	pageSize := restfulx.QueryInt(rc, "pageSize", 10)
+	ruleId := restfulx.QueryParam(rc, "ruleId")
+	nodeId := restfulx.QueryParam(rc, "nodeId")
+
+	total, list := rule_engine.GetDebugDataPage(pageNum, pageSize, ruleId, nodeId)
+	rc.ResData = model.ResultPage{
+		Total:    total,
+		PageNum:  int64(pageNum),
+		PageSize: int64(pageSize),
+		Data:     list,
+	}
 }
 
 // GetRuleChainList WorkInfo列表数据

apps/rule/entity/rulechain_data.go

@@ -5,12 +5,15 @@ import (
 )
 
 type RuleDataJson struct {
-	LfData struct {
-		GlobalColor string                 `json:"globalColor"`
-		DataCode    map[string]interface{} `json:"dataCode"`
-		OpenRule    bool                   `json:"openRule"`
-		Setting     map[string]interface{} `json:"setting"`
-	} `json:"lfData"`
+	Id     string
+	LfData LfData `json:"lfData"`
+}
+
+type LfData struct {
+	GlobalColor string                 `json:"globalColor"`
+	DataCode    map[string]interface{} `json:"dataCode"`
+	OpenRule    bool                   `json:"openRule"`
+	Setting     map[string]interface{} `json:"setting"`
 }
 
 // 序列化

apps/rule/router/rulechain.go

@@ -27,12 +27,17 @@ func InitRuleChainRouter(container *restful.Container) {
 		Metadata(restfulspec.KeyOpenAPITags, tags).
 		Returns(200, "OK", model.ResultPage{}))
 
-	ws.Route(ws.GET("/test").To(func(request *restful.Request, response *restful.Response) {
-		restfulx.NewReqCtx(request, response).WithNeedCasbin(false).WithLog("测试规则引擎").Handle(s.RuleChainTest)
+	ws.Route(ws.GET("/node/debug").To(func(request *restful.Request, response *restful.Response) {
+		restfulx.NewReqCtx(request, response).WithNeedCasbin(false).WithLog("获取规则链节点日志").Handle(s.GetNodeDebug)
 	}).
-		Doc("测试规则引擎").
-		Param(ws.QueryParameter("code", "流程代码").DataType("string")).
-		Metadata(restfulspec.KeyOpenAPITags, tags))
+		Doc("获取规则链节点日志").
+		Param(ws.QueryParameter("pageNum", "页数").Required(true).DataType("int")).
+		Param(ws.QueryParameter("pageSize", "每页条数").Required(true).DataType("int")).
+		Param(ws.QueryParameter("ruleId", "规则ID").Required(false).DataType("string")).
+		Param(ws.QueryParameter("nodeId", "节点ID").Required(false).DataType("string")).
+		Metadata(restfulspec.KeyOpenAPITags, tags).
+		Metadata(restfulspec.KeyOpenAPITags, tags).
+		Returns(200, "OK", model.ResultPage{}))
 
 	ws.Route(ws.GET("/list").To(func(request *restful.Request, response *restful.Response) {
 		restfulx.NewReqCtx(request, response).WithLog("获取规则引擎分页列表").Handle(s.GetRuleChainList)

iothub/hook_message_work/hook_message_work.go

@@ -62,7 +62,7 @@ func (s *HookService) handleOne(msg *netbase.DeviceEventInfo) {
 			dataCode := chain.LfData.DataCode
 			code, err := json.Marshal(dataCode)
 			//新建规则链实体
-			instance, errs := rule_engine.NewRuleChainInstance(code)
+			instance, errs := rule_engine.NewRuleChainInstance(chain.Id, code)
 			if len(errs) > 0 {
 				global.Log.Error("规则链初始化失败", errs[0])
 				return
@@ -116,13 +116,19 @@ func getRuleChain(etoken *global_model.DeviceAuth) *ruleEntity.RuleDataJson {
 	get, err := cache.ComputeIfAbsentProductRule(key, func(k any) (any, error) {
 		one := services.ProductModelDao.FindOne(k.(string))
 		rule := ruleService.RuleChainModelDao.FindOne(one.RuleChainId)
-		return rule.RuleDataJson, nil
+		var lfData ruleEntity.LfData
+		err := tool.StringToStruct(rule.RuleDataJson, &lfData)
+		if err != nil {
+			return nil, err
+		}
+		return ruleEntity.RuleDataJson{Id: rule.Id, LfData: lfData}, nil
 	})
 	biz.ErrIsNil(err, "缓存读取规则链失败")
-	ruleData := ruleEntity.RuleDataJson{}
-	err = tool.StringToStruct(get.(string), &ruleData)
+	if ruleData, ok := get.(ruleEntity.RuleDataJson); ok {
+		return &ruleData
+	}
 	biz.ErrIsNil(err, "规则链数据转化失败")
-	return &ruleData
+	return nil
 }
 
 // 构建规则链执行的消息

pkg/initialize/event.go

@@ -3,9 +3,11 @@ package initialize
 import (
 	"pandax/apps/device/entity"
 	"pandax/apps/device/services"
+	ruleEntity "pandax/apps/rule/entity"
 	"pandax/pkg/cache"
 	"pandax/pkg/events"
 	"pandax/pkg/global"
+	"pandax/pkg/tool"
 )
 
 // 初始化事件监听
@@ -17,8 +19,10 @@ func InitEvents() {
 			RuleChainId: ruleId,
 		})
 		if list != nil {
+			var lfData ruleEntity.LfData
+			tool.StringToStruct(codeData, &lfData)
 			for _, product := range *list {
-				cache.PutProductRule(product.Id, codeData)
+				cache.PutProductRule(product.Id, ruleEntity.RuleDataJson{Id: ruleId, LfData: lfData})
 			}
 		}
 	})

pkg/tdengine/tdengine.go

@@ -8,14 +8,6 @@ import (
 	"time"
 )
 
-const (
-	TIME_TYPE_PROP  = "telemetry"
-	TIME_TYPE_ATRE  = "attributes"
-	TIME_TYPE_LOGS  = "logs"
-	TIME_TYPE_ALARM = "alarm"
-	TIME_TYPE_EVENT = "event"
-)
-
 type TdEngine struct {
 	db     *sql.DB
 	dbName string

pkg/tdengine/tdengine_event.go

@@ -2,10 +2,10 @@ package tdengine
 
 import (
 	"fmt"
-	"github.com/kakuilan/kgo"
-	"strings"
 )
 
+const connectTableName = "device_connect"
+
 type ConnectInfo struct {
 	Ts         string `json:"ts"`
 	ClientID   string `json:"clientId"`
@@ -18,26 +18,12 @@ type ConnectInfo struct {
 
 // CreateEventTable 创建设备连接事件表
 func (s *TdEngine) CreateEventTable() (err error) {
-	sql := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.device_connect (ts TIMESTAMP,deviceId NCHAR(64),
-    type NCHAR(64),clientId NCHAR(64),peerHost NCHAR(64),sockPort NCHAR(64),protocol NCHAR(64))`, s.dbName)
+	sql := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.%s (ts TIMESTAMP,deviceId NCHAR(64),
+    type NCHAR(64),clientId NCHAR(64),peerHost NCHAR(64),sockPort NCHAR(64),protocol NCHAR(64))`, s.dbName, connectTableName)
 	_, err = s.db.Exec(sql)
 	return
 }
 
 func (s *TdEngine) InsertEvent(data map[string]any) (err error) {
-	if len(data) == 0 {
-		return
-	}
-	var (
-		field = []string{}
-		value = []string{}
-	)
-	for k, v := range data {
-		field = append(field, k)
-		value = append(value, "'"+kgo.KConv.ToStr(v)+"'")
-	}
-
-	sql := "INSERT INTO ? (?) VALUES (?)"
-	_, err = s.db.Exec(sql, "device_connect", strings.Join(field, ","), strings.Join(value, ","))
-	return err
+	return s.InsertDevice(connectTableName, data)
 }

pkg/tdengine/tdengine_log.go

@@ -2,6 +2,8 @@ package tdengine
 
 import "time"
 
+const logTableName = "device_log"
+
 // CreateLogStable 添加LOG超级表
 func (s *TdEngine) CreateLogStable() (err error) {
 	var name string

pkg/tdengine/tdengine_rule_debug.go

@@ -0,0 +1,18 @@
+package tdengine
+
+import (
+	"fmt"
+)
+
+const debugTableName = "device_rule_debug"
+
+func (s *TdEngine) CreateDeviceRuleDebugTable() (err error) {
+	sql := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.%s (ts TIMESTAMP,nodeId NCHAR(64),msgd NCHAR(64),debugType NCHAR(64),
+    deviceName NCHAR(64),msgType NCHAR(64),msg VARCHAR,metadata VARCHAR,error VARCHAR)`, s.dbName, debugTableName)
+	_, err = s.db.Exec(sql)
+	return
+}
+
+func (s *TdEngine) InsertRuleDebug(data map[string]any) (err error) {
+	return s.InsertDevice(debugTableName, data)
+}

pkg/tool/conv.go

@@ -3,6 +3,7 @@ package tool
 import (
 	"encoding/json"
 	"pandax/pkg/global"
+	"reflect"
 	"strings"
 	"time"
 )
@@ -67,6 +68,21 @@ func FirstLowCamelString(s string) string {
 	return string(data)
 }
 
+func StructToMap(s interface{}) map[string]interface{} {
+	result := make(map[string]interface{})
+
+	value := reflect.ValueOf(s)
+	typ := reflect.TypeOf(s)
+
+	for i := 0; i < value.NumField(); i++ {
+		field := typ.Field(i)
+		fieldValue := value.Field(i).Interface()
+		result[field.Name] = fieldValue
+	}
+
+	return result
+}
+
 func MapToStruct(m map[string]interface{}, s interface{}) error {
 	data, err := json.Marshal(m)
 	if err != nil {