[优化] 属性下发采用规则链rpc请求

2026-04-23 02:48:34 +08:00 · 2023-10-14 11:36:18 +08:00
parent 7c8001a687
commit 0c73dc24b9
6 changed files with 82 additions and 1240 deletions
--- a/apps/device/api/device.go
+++ b/apps/device/api/device.go
@@ -6,13 +6,11 @@ package api
 // 生成人：panda
 // ==========================================================================
 import (
-	"encoding/json"
 	"fmt"
 	"github.com/PandaXGO/PandaKit/biz"
 	"github.com/PandaXGO/PandaKit/model"
 	"github.com/PandaXGO/PandaKit/restfulx"
-	"pandax/iothub/client/mqttclient"
-	"pandax/iothub/client/tcpclient"
+	"pandax/apps/device/util"
 	"pandax/pkg/global"
 	"pandax/pkg/global_model"
 	"pandax/pkg/shadow"
@@ -147,23 +145,13 @@ func (p *DeviceApi) DownAttribute(rc *restfulx.ReqCtx) {
 	id := restfulx.PathParam(rc, "id")
 	key := restfulx.QueryParam(rc, "key")
 	value := restfulx.QueryParam(rc, "value")
-	one := p.DeviceApp.FindOne(id)
-	biz.IsTrue(one.LinkStatus == global.ONLINE, "设备不在线无法设置属性")
-	if one.Product.ProtocolName == global.TCPProtocol {
-		err := tcpclient.Send(id, value)
-		biz.ErrIsNil(err, "属性下发失败")
-	}
-	if one.Product.ProtocolName == global.MQTTProtocol {
-		// 下发指令
-		contentMap := map[string]interface{}{
+	err := util.BuildRunDeviceRpc(id, "single", map[string]interface{}{
+		"method": "setAttributes",
+		"params": map[string]interface{}{
 			key: value,
-		}
-		content, _ := json.Marshal(contentMap)
-		var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: "single"}
-		rpc.GetRequestId()
-		err := rpc.RequestAttributes(global_model.RpcPayload{Params: string(content)})
-		biz.ErrIsNil(err, "属性下发失败")
-	}
+		},
+	})
+	biz.ErrIsNilAppendErr(err, "下发失败:")
 }

 // InsertDevice 添加Device
--- a/apps/device/api/device_cmd.go
+++ b/apps/device/api/device_cmd.go
@@ -2,18 +2,13 @@ package api

 // ==========================================================================
 import (
-	"context"
 	"encoding/json"
 	"github.com/PandaXGO/PandaKit/biz"
 	"github.com/PandaXGO/PandaKit/model"
 	"github.com/PandaXGO/PandaKit/restfulx"
-	ruleEntity "pandax/apps/rule/entity"
-	ruleService "pandax/apps/rule/services"
+	"pandax/apps/device/util"
 	"pandax/pkg/global"
 	"pandax/pkg/global_model"
-	"pandax/pkg/rule_engine"
-	"pandax/pkg/rule_engine/message"
-	"pandax/pkg/tool"
 	"strings"
 	"time"

@@ -53,44 +48,16 @@ func (p *DeviceCmdLogApi) InsertDeviceCmdLog(rc *restfulx.ReqCtx) {
 	ms := make(map[string]interface{})
 	err := json.Unmarshal([]byte(data.CmdContent), &ms)
 	biz.ErrIsNil(err, "指令格式不正确")
-
 	data.Id = global_model.GenerateID()
 	data.State = "2"
 	data.RequestTime = time.Now().Format("2006-01-02 15:04:05")
-	one := p.DeviceApp.FindOne(data.DeviceId)
-	biz.IsTrue(one.LinkStatus == global.ONLINE, "设备不在线无法下发指令")
-	// 查询规则链
-	findOne := ruleService.RuleChainModelDao.FindOne(one.Product.RuleChainId)
-	ruleData := ruleEntity.RuleDataJson{}
-	err = tool.StringToStruct(findOne.RuleDataJson, &ruleData)
-	biz.ErrIsNil(err, "规则链数据转化失败")
-	dataCode := ruleData.LfData.DataCode
-	code, err := json.Marshal(dataCode)
-	//新建规则链实体
-	instance, errs := rule_engine.NewRuleChainInstance(code)
-	if len(errs) > 0 {
-		global.Log.Error("规则链初始化失败", errs[0])
-		return
-	}
 	go func() {
-		// 构建规则链消息
-		metadataVals := map[string]interface{}{
-			"deviceId":       data.DeviceId,
-			"mode":           data.Mode,
-			"deviceName":     one.Name,
-			"deviceType":     one.DeviceType,
-			"deviceProtocol": one.Product.ProtocolName,
-			"productId":      one.Pid,
-			"orgId":          one.OrgId,
-			"owner":          one.Owner,
-		}
-		msg := message.NewMessage(one.Owner, message.RpcRequestToDevice, map[string]interface{}{
+		err := util.BuildRunDeviceRpc(data.DeviceId, data.Mode, map[string]interface{}{
 			"method": data.CmdName,
 			"params": ms,
-		}, metadataVals)
-		err = instance.StartRuleChain(context.Background(), msg)
+		})
 		if err != nil {
-			global.Log.Error("规则链执行失败", errs)
+			global.Log.Error("规则链执行失败", err)
 			data.State = "1"
 		} else {
 			data.State = "0"
@@ -99,7 +66,6 @@ func (p *DeviceCmdLogApi) InsertDeviceCmdLog(rc *restfulx.ReqCtx) {
 		err = p.DeviceCmdLogApp.Insert(data)
 		biz.ErrIsNil(err, "添加指令记录失败")
 	}()
-
 }

 // DeleteDeviceCmdLog 删除告警
--- a/apps/device/util/device_rpc.go
+++ b/apps/device/util/device_rpc.go
@@ -0,0 +1,51 @@
+package util
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"pandax/apps/device/services"
+	ruleEntity "pandax/apps/rule/entity"
+	ruleService "pandax/apps/rule/services"
+	"pandax/pkg/global"
+	"pandax/pkg/rule_engine"
+	"pandax/pkg/rule_engine/message"
+	"pandax/pkg/tool"
+)
+
+func BuildRunDeviceRpc(deviceId, mode string, metadata map[string]interface{}) error {
+	one := services.DeviceModelDao.FindOne(deviceId)
+	if one.LinkStatus != global.ONLINE {
+		return errors.New("设备不在线无法设置属性")
+	}
+	findOne := ruleService.RuleChainModelDao.FindOne(one.Product.RuleChainId)
+	ruleData := ruleEntity.RuleDataJson{}
+	err := tool.StringToStruct(findOne.RuleDataJson, &ruleData)
+	if err != nil {
+		global.Log.Error("规则链数据转化失败", err)
+		return errors.New("规则链数据转化失败")
+	}
+	dataCode := ruleData.LfData.DataCode
+	code, _ := json.Marshal(dataCode)
+	//新建规则链实体
+	instance, errs := rule_engine.NewRuleChainInstance(code)
+	if len(errs) > 0 {
+		return errs[0]
+	}
+	metadataVals := map[string]interface{}{
+		"deviceId":       one.Id,
+		"mode":           mode,
+		"deviceName":     one.Name,
+		"deviceType":     one.DeviceType,
+		"deviceProtocol": one.Product.ProtocolName,
+		"productId":      one.Pid,
+		"orgId":          one.OrgId,
+		"owner":          one.Owner,
+	}
+	msg := message.NewMessage(one.Owner, message.RpcRequestToDevice, metadata, metadataVals)
+	err = instance.StartRuleChain(context.Background(), msg)
+	if err != nil {
+		global.Log.Error("规则链执行失败", errs)
+	}
+	return err
+}
--- a/fatal/latest.log
+++ b/fatal/latest.log
--- a/pkg/websocket/socket_server.go
+++ b/pkg/websocket/socket_server.go
@@ -6,9 +6,8 @@ import (
 	"github.com/gorilla/websocket"
 	"net/http"
 	"pandax/apps/device/entity"
-	"pandax/iothub/client/mqttclient"
+	"pandax/apps/device/util"
 	"pandax/pkg/global"
-	"pandax/pkg/global_model"
 	"strings"
 )

@@ -40,18 +39,18 @@ func NewWebsocket(writer http.ResponseWriter, r *http.Request, header http.Heade

 // OnMessage 消息
 // 发送消息消息类型 01:发送的设备数据  02:收到指令回复  03: 心跳回复
-func OnMessage(ws *Websocket, message string) {
-	if message != "" && strings.Index(message, "ONLINE") != -1 {
-		screenId := strings.Split(message, "ONLINE")[0]
+func OnMessage(ws *Websocket, msg string) {
+	if msg != "" && strings.Index(msg, "ONLINE") != -1 {
+		screenId := strings.Split(msg, "ONLINE")[0]
 		AddWebSocketByScreenId(screenId, ws)
 	}
 	//画布离开
-	if message != "" && strings.Index(message, "LEAVE") != -1 {
-		RemoveWebSocket(strings.Split(message, "LEAVE")[0])
+	if msg != "" && strings.Index(msg, "LEAVE") != -1 {
+		RemoveWebSocket(strings.Split(msg, "LEAVE")[0])
 	}
 	//客户端传来了控制命令 格式   场景控制代码CONTROLCMD控制命令CONTROLCMD传感器id
-	if message != "" && strings.Index(message, "CONTROLCMD") != -1 {
-		split := strings.Split(message, "CONTROLCMD")
+	if msg != "" && strings.Index(msg, "CONTROLCMD") != -1 {
+		split := strings.Split(msg, "CONTROLCMD")
 		if len(split) < 2 {
 			return
 		}
@@ -62,25 +61,24 @@ func OnMessage(ws *Websocket, message string) {
 		err := json.Unmarshal([]byte(controlCMD), vtsa)
 		if err != nil {
 			global.Log.Error("设备参数下发，孪生体参数解析失败", err)
-			sendMessages("02", "下发失败", screenId)
+			sendMessages("02", "下发失败,设备参数结构错误", screenId)
 			return
 		}
 		//2. 根据设备下发属性更改
-		//topic := fmt.Sprintf(global.AttributesTopic, vtsa.TwinId)
-		content, _ := json.Marshal(vtsa.Attrs)
-		var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: "single"}
-		rpc.GetRequestId()
-		err = rpc.RequestAttributes(global_model.RpcPayload{Params: string(content)})
+		err = util.BuildRunDeviceRpc(vtsa.TwinId, "single", map[string]interface{}{
+			"method": "setAttributes",
+			"params": vtsa.Attrs,
+		})
 		if err != nil {
-			global.Log.Error("属性下发失败", err)
-			sendMessages("02", "下发失败", screenId)
-			return
-		}
+			global.Log.Error("命令发送失败", err)
+			sendMessages("02", "命令发送失败", screenId)
+		} else {
 			sendMessages("02", "命令发送成功", screenId)
 		}
+	}
 	//心跳处理
-	if message != "" && strings.Index(message, "HEARTCMD") != -1 {
-		split := strings.Split(message, "HEARTCMD")
+	if msg != "" && strings.Index(msg, "HEARTCMD") != -1 {
+		split := strings.Split(msg, "HEARTCMD")
 		if len(split) < 1 {
 			return
 		}
--- a/shutdown.bat
+++ b/shutdown.bat
@@ -1 +1 @@
-taskkill /pid 28744  -t  -f
+taskkill /pid 31740  -t  -f