[feat]udp协议数据上传解析

apps/device/api/device.go

@@ -105,20 +105,37 @@ func (p *DeviceApi) GetDeviceStatus(rc *restfulx.ReqCtx) {
 		rs = getDevice.TelemetryPoints
 	}
 	for _, tel := range *template {
+		if _, ok := rs[tel.Key]; !ok {
+			continue
+		}
 		sdv := entity.DeviceStatusVo{
 			Name:   tel.Name,
 			Key:    tel.Key,
 			Type:   tel.Type,
 			Define: tel.Define,
 		}
-		if v, ok := rs[tel.Key]; ok {
-			sdv.Value = v.Value
-			sdv.Time = v.UpdatedAt
-		} else {
-			if classify == global.TslAttributesType {
-				if value, ok := tel.Define["default_value"]; ok {
-					sdv.Value = value
+		if classify == global.TslTelemetryType {
+			value := rs[tel.Key].Value
+			// tsl转化
+			/*var tslValue tsl.ValueType
+			err := tool.MapToStruct(tel.Define, &tslValue)
+			if err != nil {
+				value = rs[tel.Key].Value
+			} else {
+				tslValue.Type = tel.Type
+				// 此处rs[tel.Key].Value 变成字符串类型了
+				value = tslValue.ConvertValue(rs[tel.Key].Value)
+				log.Println("value", value)
+				if value == nil {
+					value = rs[tel.Key]
 				}
+			}*/
+			sdv.Time = rs[tel.Key].UpdatedAt
+			sdv.Value = value
+		}
+		if classify == global.TslAttributesType {
+			if value, ok := tel.Define["default_value"]; ok {
+				sdv.Value = value
 			}
 		}
 		res = append(res, sdv)

iothub/client/updclient/udp.go

@@ -1,4 +1,4 @@
-package tcpclient
+package udpclient
 
 import (
 	"encoding/hex"
@@ -6,12 +6,17 @@ import (
 	"pandax/pkg/global"
 )
 
-var UdpClient = make(map[string]*net.UDPConn)
+type UdpClientT struct {
+	Conn *net.UDPConn
+	Addr *net.UDPAddr
+}
+
+var UdpClient = make(map[string]*UdpClientT)
 
 func Send(deviceId, msg string) error {
 	if conn, ok := UdpClient[deviceId]; ok {
 		global.Log.Infof("设备%s, 发送指令%s", deviceId, msg)
-		_, err := conn.Write([]byte(msg))
+		_, err := conn.Conn.WriteToUDP([]byte(msg), conn.Addr)
 		if err != nil {
 			return err
 		}
@@ -28,7 +33,7 @@ func SendHex(deviceId, msg string) error {
 		if err != nil {
 			return err
 		}
-		_, err = conn.Write(b)
+		_, err = conn.Conn.WriteToUDP(b, conn.Addr)
 		if err != nil {
 			return err
 		}

iothub/hook_message_work/hook_message_work.go

@@ -5,9 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/PandaXGO/PandaKit/biz"
-	"pandax/apps/device/entity"
 	"pandax/apps/device/services"
-	"pandax/apps/device/tsl"
 	ruleEntity "pandax/apps/rule/entity"
 	ruleService "pandax/apps/rule/services"
 	"pandax/iothub/netbase"
@@ -19,7 +17,6 @@ import (
 	"pandax/pkg/shadow"
 	"pandax/pkg/tool"
 	"pandax/pkg/websocket"
-	"strings"
 )
 
 // 消息处理模块
@@ -169,30 +166,13 @@ func SetDeviceShadow(etoken *global_model.DeviceAuth, msgVals map[string]interfa
 	if msgType == message.RowMes {
 		msgType = message.TelemetryMes
 	}
-	template := services.ProductTemplateModelDao.FindList(entity.ProductTemplate{Classify: strings.ToLower(msgType), Pid: etoken.ProductId})
-	for _, tel := range *template {
-		if _, ok := msgVals[tel.Key]; !ok {
-			continue
-		}
+	for key, value := range msgVals {
 		if message.AttributesMes == msgType {
-			err := shadow.DeviceShadowInstance.SetDevicePoint(etoken.Name, global.TslAttributesType, tel.Key, msgVals[tel.Key])
+			err := shadow.DeviceShadowInstance.SetDevicePoint(etoken.Name, global.TslAttributesType, key, value)
 			biz.ErrIsNilAppendErr(err, "设置设备影子点失败")
 		}
 		if message.TelemetryMes == msgType {
-			var value interface{}
-			// tsl转化
-			var tslValue tsl.ValueType
-			err := tool.MapToStruct(tel.Define, &tslValue)
-			if err != nil {
-				value = msgVals[tel.Key]
-			} else {
-				tslValue.Type = tel.Type
-				value = tslValue.ConvertValue(msgVals[tel.Key])
-				if value == nil {
-					value = msgVals[tel.Key]
-				}
-			}
-			err = shadow.DeviceShadowInstance.SetDevicePoint(etoken.Name, global.TslTelemetryType, tel.Key, value)
+			err := shadow.DeviceShadowInstance.SetDevicePoint(etoken.Name, global.TslTelemetryType, key, value)
 			biz.ErrIsNilAppendErr(err, "设置设备影子点失败")
 		}
 	}

iothub/server/emqxserver/hook.go

@@ -176,7 +176,7 @@ func (s *HookGrpcService) OnMessagePublish(ctx context.Context, in *exhook2.Mess
 	res.Type = exhook2.ValuedResponse_STOP_AND_RETURN
 	res.Value = &exhook2.ValuedResponse_BoolResult{BoolResult: false}
 	//服务端的HTTP请求放行
-	if in.GetMessage().From == "http_api" {
+	if in.Message.From == "http_api" {
 		res.Value = &exhook2.ValuedResponse_BoolResult{BoolResult: true}
 		return res, nil
 	}
@@ -193,7 +193,6 @@ func (s *HookGrpcService) OnMessagePublish(ctx context.Context, in *exhook2.Mess
 		DeviceId:   etoken.DeviceId,
 		DeviceAuth: etoken,
 	}
-
 	// 如果是网关子设备单独处理
 	if eventType == message.GATEWAY {
 		subData := make(map[string]interface{})

iothub/server/udpserver/hook.go

@@ -3,17 +3,20 @@ package updserver
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
 	"net"
+	udpclient "pandax/iothub/client/updclient"
 	"pandax/iothub/hook_message_work"
 	"pandax/iothub/netbase"
 	"pandax/pkg/global"
+	"pandax/pkg/global_model"
 	"pandax/pkg/rule_engine/message"
+	"time"
 )
 
 type HookUdpService struct {
 	HookService *hook_message_work.HookService
 	conn        *net.UDPConn
-	addr        *net.UDPAddr
 }
 
 func InitUdpHook(addr string, hs *hook_message_work.HookService) {
@@ -26,40 +29,76 @@ func InitUdpHook(addr string, hs *hook_message_work.HookService) {
 		global.Log.Infof("UDP IOTHUB HOOK Start SUCCESS, Server listen: %s", addr)
 	}
 	buffer := make([]byte, 1024)
+	isAuth := false
+	etoken := &global_model.DeviceAuth{}
+	hhs := &HookUdpService{
+		HookService: hs,
+		conn:        server.listener,
+	}
 	for {
 		n, client, err := server.listener.ReadFromUDP(buffer)
 		if err != nil {
 			global.Log.Error("Error accepting connection:", err)
+			_ = server.listener.Close()
+			//设置断开连接
+			if isAuth {
+				data := netbase.CreateConnectionInfo(message.DisConnectMes, "udp", client.IP.String(), client.AddrPort().String(), etoken)
+				hhs.HookService.MessageCh <- data
+			}
+			delete(udpclient.UdpClient, etoken.DeviceId)
+			isAuth = false
 			continue
 		}
-		hhs := &HookUdpService{
-			HookService: hs,
-			conn:        server.listener,
-			addr:        client,
-		}
-		go hhs.hook(buffer[:n])
+		if !isAuth {
+			token := string(buffer[:n])
+			etoken.GetDeviceToken(token)
+			auth := netbase.Auth(token)
+			// 认证成功，创建连接记录
+			if auth {
+				global.Log.Infof("UDP协议 设备%s,认证成功", etoken.DeviceId)
+				data := netbase.CreateConnectionInfo(message.ConnectMes, "udp", client.IP.String(), client.AddrPort().String(), etoken)
+				hhs.HookService.MessageCh <- data
+				isAuth = true
+				udpclient.UdpClient[etoken.DeviceId] = &udpclient.UdpClientT{
+					Conn: server.listener,
+					Addr: client,
+				}
+				hhs.Send(client, "success")
+			} else {
+				hhs.Send(client, "fail")
+			}
+		} else {
+			data := &netbase.DeviceEventInfo{
+				DeviceId:   etoken.DeviceId,
+				DeviceAuth: etoken,
+			}
 
+			hexData := hex.EncodeToString(buffer[:n])
+			global.Log.Infof("UDP协议 设备%s, 接受消息%s", etoken.DeviceId, hexData)
+			ts := time.Now().Format("2006-01-02 15:04:05.000")
+			data.Type = message.RowMes
+			data.Datas = fmt.Sprintf(`{"ts": "%s","rowdata": "%s"}`, ts, hexData)
+
+			// etoken中添加设备标识
+			hhs.HookService.MessageCh <- data
+		}
 	}
 }
 
-func (hhs *HookUdpService) hook(data []byte) {
-	hhs.Send("success")
+func (hhs *HookUdpService) Send(addr *net.UDPAddr, message string) error {
+	return hhs.SendBytes(addr, []byte(message))
 }
 
-func (hhs *HookUdpService) Send(message string) error {
-	return hhs.SendBytes([]byte(message))
-}
-
-func (hhs *HookUdpService) SendHex(msg string) error {
+func (hhs *HookUdpService) SendHex(addr *net.UDPAddr, msg string) error {
 	b, err := hex.DecodeString(msg)
 	if err != nil {
 		return err
 	}
-	return hhs.SendBytes(b)
+	return hhs.SendBytes(addr, b)
 }
 
-func (hhs *HookUdpService) SendBytes(msg []byte) error {
-	_, err := hhs.conn.WriteToUDP(msg, hhs.addr)
+func (hhs *HookUdpService) SendBytes(addr *net.UDPAddr, msg []byte) error {
+	_, err := hhs.conn.WriteToUDP(msg, addr)
 	if err != nil {
 		hhs.conn.Close()
 		data := &netbase.DeviceEventInfo{