API接入

Handle/Handle.go

@@ -1,35 +1,212 @@
 package Handle
 
 import (
+	"Yunlot/conf"
+	"Yunlot/lib"
 	"Yunlot/logs"
 	"Yunlot/models/Device"
 	"Yunlot/models/Product"
+	"encoding/json"
 	"fmt"
 	"go.mongodb.org/mongo-driver/bson"
+	"plugin"
 	"reflect"
 )
 
-func AnalysisMap(Device_r *Device.Device, ProductType_r Product.ProductType, ArticleSlide map[string]interface{}, JointTab string) {
+// 设备->平台
+func PullHandle(Device_r *Device.Device, topicName string, message []byte) lib.JSONR {
+	var Rt_r = lib.JSONR{Code: 200, Msg: "ok"}
+
+	// 设备协议
+	ProductProt_r := Product.ProductProt{Id: Device_r.T_ProductJson.T_prot}
+	if !ProductProt_r.Read() {
+		logs.Println("MqttServer", Device_r.T_sn+"|"+Device_r.T_ProductID+"-"+fmt.Sprintf("%d", Device_r.T_ProductJson.T_prot)+" 设备协议找不到！")
+		Rt_r.Code = 203
+		Rt_r.Msg = "T_prot E!"
+		return Rt_r
+	}
+
+	messagejson := string(message)
+	// 是否加载转换协议
+	if ProductProt_r.T_lang != 0 && len(ProductProt_r.T_analysis) != 0 {
+		// 加载 SO 文件
+		p, err := plugin.Open(conf.Analysis_Dir + ProductProt_r.T_analysis + ".so")
+		if err != nil {
+			logs.PrintlnError("打开 SO 失败：", err)
+			Rt_r.Code = 203
+			Rt_r.Msg = "T_analysis E!"
+			return Rt_r
+		}
+		logs.Println("Plugin 地址：", &p)
+
+		// 查找库导出信息
+		s, err := p.Lookup("T")
+		if err != nil {
+			logs.Println("", err)
+			panic(any(err))
+		}
+		// 类型转换
+		messagejson = s.(func(t string, b []byte) string)(topicName, message)
+		// 开始处理
+		logs.Println("协议后：", messagejson)
+
+	}
+
+	//logs.Println("首字符：", string(f[0]))
+	if messagejson[0] == '{' {
+		//结构体
+		var json_r map[string]interface{}
+		err := json.Unmarshal([]byte(messagejson), &json_r)
+		if err != nil {
+			Rt_r.Code = 203
+			Rt_r.Msg = "json E!"
+			return Rt_r
+		}
+
+		json_r = AnalysisMap(Device_r, json_r, "")
+		logs.Println("json_r:", json_r)
+
+		// 合并json
+		Device_r.Read_Tidy() // 提前最新的
+		logs.Println("Device_r.T_data:", Device_r.T_data)
+		var json_x map[string]interface{}
+		if len(Device_r.T_data) > 5 {
+			json.Unmarshal([]byte(Device_r.T_data), &json_x)
+			json_x_b, _ := json.Marshal(json_r)
+			json.Unmarshal(json_x_b, &json_x)
+			json_r = json_x
+		}
+
+		logs.Println("Device_r.T_data 更新:", json_r)
+		Device_r.T_dataJson = json_r
+		Device_r.UpdateTime.NowDbTime()
+		Device_r.Update("T_data", "UpdateTime")
+
+	}
+	//
+	//else {
+	//	//列表
+	//	var json_lr []map[string]interface{}
+	//	err := json.Unmarshal([]byte(messagejson), &json_lr)
+	//	if err != nil {
+	//		Rt_r.Code = 204
+	//		Rt_r.Msg = "[]json E!"
+	//		return Rt_r
+	//	}
+	//	for _, value := range json_lr {
+	//		json_lr[len(json_lr)-1] = AnalysisMap(Device_r, value, "")
+	//	}
+	//
+	//	// 合并json
+	//	Device_r.Read_Tidy() // 提前最新的
+	//	logs.Println("Device_r.T_data:", Device_r.T_data)
+	//	var json_x map[string]interface{}
+	//	if len(Device_r.T_data) > 5 {
+	//		json.Unmarshal([]byte(Device_r.T_data), &json_x)
+	//		json_x_b, _ := json.Marshal(json_lr[len(json_lr)-1])
+	//		json.Unmarshal(json_x_b, &json_x)
+	//		json_lr[len(json_lr)-1] = json_x
+	//	}
+	//
+	//	Device_r.T_dataJson = json_lr[len(json_lr)-1]
+	//	Device_r.UpdateTime.NowDbTime()
+	//	Device_r.Update("T_data", "UpdateTime")
+	//}
+
+	return Rt_r
+}
+
+// 平台->设备
+func PushHandle(Device_r *Device.Device, topicName string, message string) (string, []byte) {
+
+	// 设备协议
+	ProductProt_r := Product.ProductProt{Id: Device_r.T_ProductJson.T_prot}
+	if !ProductProt_r.Read() {
+		logs.Println("PushHandle", Device_r.T_sn+"|"+Device_r.T_ProductID+"-"+fmt.Sprintf("%d", Device_r.T_ProductJson.T_prot)+" 设备协议找不到！")
+		return "", []byte{}
+	}
+
+	var byte_r []byte
+	var topicName_r = topicName
+	// 是否加载转换协议
+	if ProductProt_r.T_lang != 0 && len(ProductProt_r.T_analysis) != 0 {
+		// 根据库的存放路径加载库
+		p, err := plugin.Open(conf.Analysis_Dir + ProductProt_r.T_analysis + ".so")
+		if err != nil {
+			logs.PrintlnError("PushHandle:", err)
+			return "", []byte{}
+		}
+
+		// 查找库导出信息
+		s, err := p.Lookup("R")
+		if err != nil {
+			logs.PrintlnError("PushHandle:", err)
+			return "", []byte{}
+		}
+		// 类型转换
+		topicName_r, byte_r = s.(func(sn string, b string) (string, []byte))(topicName, message)
+
+	}
+
+	// 无效消息，不用推送
+	if len(topicName_r) == 0 || len(byte_r) == 0 {
+		return "", []byte{}
+	}
+
+	// 长连接 网关
+	switch ProductProt_r.T_mode {
+	case 1: //mqtt
+		// 如果 订阅地址与发布相同，在后面强行加 _reply，避免发布后无法收到消息
+		if topicName_r == topicName {
+			topicName_r += "_reply"
+		}
+		lib.Mqtt_publish(topicName_r, byte_r) // 返回数据
+		break
+		//case 2: //tcp
+		//
+		//	break
+		//case 3: //CoAP
+		//
+		//	break
+		//case 4: //websocket
+		//
+		//	break
+
+	}
+
+	return topicName_r, byte_r
+
+}
+
+// 处理数据
+func AnalysisMap(Device_r *Device.Device, ArticleSlide map[string]interface{}, JointTab string) (json_r map[string]interface{}) {
 	//JointTab += "."
 	T_r := make(map[string]interface{})
+	json_r = make(map[string]interface{})
 	for key, value := range ArticleSlide {
 		//fmt.Println(reflect.TypeOf(value).String())
 		switch reflect.TypeOf(value).String() {
 		case "map[string]interface {}":
-			AnalysisMap(Device_r, ProductType_r, value.(map[string]interface{}), JointTab+key+".")
+			json_r[key] = AnalysisMap(Device_r, value.(map[string]interface{}), JointTab+key+".")
+			return json_r
 			break
 		case "[]interface {}":
 			for _, valuex := range value.([]interface{}) {
 				if reflect.TypeOf(valuex).String() == "map[string]interface {}" {
-					AnalysisMap(Device_r, ProductType_r, valuex.(map[string]interface{}), JointTab+key+".")
+					json_r[key] = AnalysisMap(Device_r, valuex.(map[string]interface{}), JointTab+key+".")
 				}
 			}
+			return json_r
 			break
 		default:
 			T_r[key] = value
 			// 根 TAB
 			if len(JointTab) == 0 {
-				Device.Data_Add(Device_r.T_sn+"_"+key, &bson.M{key: value})
+				//logs.Println("TAB.:",key)
+				bson_r := bson.M{key: value}
+				go Device.Data_Add(Device_r.T_sn+"_"+key, &bson_r)
+				// 消息转发
+				go Relay(Device_r, JointTab, &bson_r)
 			}
 
 			break
@@ -40,73 +217,18 @@ func AnalysisMap(Device_r *Device.Device, ProductType_r Product.ProductType, Art
 	}
 	// 多级 TAB
 	JointTab = JointTab[:len(JointTab)-1]
-	fmt.Println(JointTab, "-------")
+	//logs.Println("TAB-:",JointTab)
 	bson_r := bson.M{}
 	for key, value := range T_r {
 		fmt.Println(key, "->:", value)
 		bson_r[key] = value
+		json_r[key] = value
 	}
 
-	// 消息转发
-	//Relay(Device_r, ProductType_r, JointTab, ArticleSlide)
-
 	// 数据存储
-	Device.Data_Add(Device_r.T_sn+"_"+JointTab, &bson_r)
-
-}
-
-// 设备->平台
-func T(Device_r Device.Device, topic string, message []byte) {
-	// 设备类型
-	ProductType_r := Product.ProductType{T_ProductID: Device_r.T_ProductID}
-	if !ProductType_r.Read() {
-		logs.Println("MqttServer", Device_r.T_sn+"|"+Device_r.T_ProductID+" 设备类型找不到！")
-		return
-	}
-
-	// 设备协议
-	ProductProt_r := Product.ProductProt{Id: ProductType_r.T_prot}
-	if !ProductProt_r.Read() {
-		logs.Println("MqttServer", Device_r.T_sn+"|"+Device_r.T_ProductID+"-"+fmt.Sprintf("%d", ProductType_r.T_prot)+" 设备协议找不到！")
-		return
-	}
-	//
-	//// 根据库的存放路径加载库
-	//p, err := plugin.Open(conf.Analysis_Dir + ProductProt_r.T_analysis)
-	//if err != nil {
-	//	println(err)
-	//	panic(any(err))
-	//}
-	//
-	//// 查找库导出信息
-	//s, err := p.Lookup("T")
-	//if err != nil {
-	//	println(err)
-	//	panic(any(err))
-	//}
-	//// 类型转换
-	//f := s.(func(b []byte) []byte)(message)
-
-	////0：Mqtt  1：http  2:tcp  3:CoAP  4:websocket
-	//switch ProductProt_r.T_mode {
-	//case 0: //Mqtt
-	//	MqttServer.Mqtt_publish(topic, f) // 返回数据
-	//	break
-	//case 1: //http
-	//
-	//	break
-	//case 2: //tcp
-	//
-	//	break
-	//case 3: //CoAP
-	//
-	//	break
-	//case 4: //websocket
-	//
-	//	break
-	//
-	//}
-
-	return
+	go Device.Data_Add(Device_r.T_sn+"_"+JointTab, &bson_r)
 
+	// 消息转发
+	go Relay(Device_r, JointTab, &bson_r)
+	return json_r
 }

Handle/HttpServer/Server.go

@@ -1,46 +1,114 @@
 package HttpServer
 
 import (
+	"Yunlot/Handle"
 	"Yunlot/conf"
+	"Yunlot/lib"
+	"Yunlot/logs"
+	"Yunlot/models/Device"
+	"encoding/json"
 	"fmt"
 	"github.com/gin-gonic/gin"
+	"strings"
 )
 
+func ReturnHandle(Device_r *Device.Device, topicName string, data lib.JSONR) interface{} {
+	b, _ := json.Marshal(data)
+	_, byte_r := Handle.PushHandle(Device_r, topicName, string(b))
+	var json_r map[string]interface{}
 
+	if len(byte_r) > 0 {
+		err := json.Unmarshal(byte_r, &json_r)
+		if err == nil {
+			return json_r
+		}
+		return byte_r
+	}
+
+	err := json.Unmarshal(byte_r, &json_r)
+	if err == nil {
+		return json_r
+	}
+	return b
+}
+
+func json_key() {
+
+}
 
 func Run() {
 
 	// 创建一个Gin路由器
 	r := gin.Default()
 
-	//设备上线
-	r.POST("/online", func(c *gin.Context) {
+	//设备上报属性
+	r.POST("/*topicName", func(c *gin.Context) {
+		logs.Println("T_sn:", c.Query("T_sn"))
+		logs.Println("topicName:", c.Param("topicName"))
 
-		c.JSON(200, gin.H{"result": "allow"})
-		return
-	})
-	//设备离线
-	r.POST("/offline", func(c *gin.Context) {
+		Device_r := Device.Device{T_sn: c.Query("T_sn")}
+		if !Device_r.Read_Tidy() {
+			c.JSON(200, lib.JSONR{Code: lib.Error, Msg: "T_sn 错误！"})
+			return
+		}
+		if Device_r.T_state != 1 {
+			c.JSON(200, lib.JSONR{Code: lib.Error, Msg: "T_sn 错误！"})
+			return
+		}
+		if Device_r.T_password != c.Query("T_password") {
+			c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), lib.JSONR{Code: lib.Error, Msg: "T_password 错误！"}))
+			return
+		}
 
-		c.JSON(200, gin.H{"result": "allow"})
-		return
-	})
+		//
+		byte_r, err := c.GetRawData()
+		if err != nil {
+			c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), lib.JSONR{Code: lib.Error, Msg: "JSON 错误！"}))
+			return
+		}
+		if len(byte_r) == 0 {
+			c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), lib.JSONR{Code: lib.Error, Msg: "JSON 错误！"}))
+			return
+		}
+		logs.Println("byte", string(byte_r))
 
-	//设备上报属性
-	r.POST("/report", func(c *gin.Context) {
+		// 转换
+		Rt_r := Handle.PullHandle(&Device_r, c.Param("topicName"), byte_r)
+		//logs.Println("ReturnHandle:",string(ReturnHandle(&Device_r,c.Param("topicName"),Rt_r)))
 
-		c.JSON(200, gin.H{"result": "allow"})
+		c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), Rt_r))
 		return
 	})
 
 	//获取属性
-	r.POST("/get", func(c *gin.Context) {
+	r.GET("/*topicName", func(c *gin.Context) {
+		logs.Println("T_sn:", c.Query("T_sn"))
+		logs.Println("topicName:", c.Param("topicName"))
+
+		Device_r := Device.Device{T_sn: c.Query("T_sn")}
+		if !Device_r.Read_Tidy() {
+			c.JSON(200, lib.JSONR{Code: lib.Error, Msg: "T_sn 错误！"})
+			return
+		}
+		if Device_r.T_state != 1 {
+			c.JSON(200, lib.JSONR{Code: lib.Error, Msg: "T_sn 错误！"})
+			return
+		}
+		if Device_r.T_password != c.Query("T_password") {
+			c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), lib.JSONR{Code: lib.Error, Msg: "T_password 错误！"}))
+			return
+		}
 
-		c.JSON(200, gin.H{"result": "allow"})
+		topicName := strings.Trim(c.Param("topicName"), "/") // 参数： /A/B/C/D
+
+		topicNameList := strings.Split(topicName, "/") // 分割数组
+
+		x := lib.Json_key(topicNameList, 0, Device_r.T_dataJson) //获取指定 json 数据，如果失败，返回已找到的全部json
+
+		c.JSON(200, ReturnHandle(&Device_r, c.Param("topicName"), lib.JSONR{Code: lib.Success, Msg: "ok！", Data: x}))
 		return
 	})
 
-
-	r.Run( fmt.Sprintf(":%d",conf.HTTPServer_Port))
+	r.Run(fmt.Sprintf(":%d", conf.HTTPServer_Port))
 
 }

Handle/MqttServer/Clients.go

@@ -1,9 +1,11 @@
 package MqttServer
 
 import (
+	"Yunlot/Handle"
 	"Yunlot/logs"
 	"Yunlot/models/Device"
 	"encoding/json"
+	"time"
 )
 
 // /  -   连接 实体
@@ -54,10 +56,17 @@ func MessageConnected(topicName string, message []byte) {
 	logs.Println("Clientid:", Ms_project.Clientid)
 	logs.Println("Username:", Ms_project.Username)
 
-	r_Device.T_state = 1
 	r_Device.T_online = 1
 	// 同步参数
-	r_Device.Update("T_state","T_online")
+	r_Device.Update("T_online")
+
+	var Device_online_r map[string]interface{}
+	Device_online_r = make(map[string]interface{})
+	Device_online_r["online"] = r_Device.T_online
+	Device_online_r["msg"] = "设备主动上线[" + Ms_project.Ipaddress + "]"
+	Device_online_r["time"] = time.Now().Format("2006-01-02 15:04:05")
+
+	Handle.AnalysisMap(&r_Device, map[string]interface{}{"online": Device_online_r}, "")
 
 	logs.Println("============= MessageConnected Mqtt JSON AND =============", topicName)
 }
@@ -112,10 +121,18 @@ func MessageDisconnected(topicName string, message []byte) {
 		return
 	}
 
-	r_Device.T_online = 0
+	r_Device.T_online = 2
 	// 同步参数
 	r_Device.Update("T_online")
 
+	var Device_online_r map[string]interface{}
+	Device_online_r = make(map[string]interface{})
+	Device_online_r["online"] = r_Device.T_online
+	Device_online_r["msg"] = "设备离线[" + ToReason(Ms_project.Reason) + "]"
+	Device_online_r["time"] = time.Now().Format("2006-01-02 15:04:05")
+
+	Handle.AnalysisMap(&r_Device, map[string]interface{}{"online": Device_online_r}, "")
+
 	logs.Println("============= MessageDisconnected Mqtt JSON AND =============", topicName)
 
 }

Handle/MqttServer/Server.go

@@ -6,28 +6,24 @@ import (
 	"Yunlot/lib"
 	"Yunlot/logs"
 	"Yunlot/models/Device"
-	"Yunlot/models/Product"
 	"encoding/json"
 	"fmt"
 	beego "github.com/beego/beego/v2/server/web"
 	"github.com/gin-gonic/gin"
 	"github.com/yosssi/gmq/mqtt"
 	"github.com/yosssi/gmq/mqtt/client"
-	"plugin"
 	"strconv"
 	"strings"
 	"time"
 )
 
-var cli *client.Client
-
 func Run_MqttServer() {
 
 	time.Sleep(3 * time.Second)
 	logs.Println("============Run_MqttServer=============", "")
 	HTTPPort, _ := beego.AppConfig.String("HTTPPort")
 	// Create an MQTT Client.
-	cli = client.New(&client.Options{
+	lib.MqttClient = client.New(&client.Options{
 		// Define the processing of the error handler.
 		ErrorHandler: func(err error) {
 			logs.PrintlnError("err!!!!!! Run_MqttServer:", err.Error())
@@ -40,7 +36,7 @@ func Run_MqttServer() {
 	})
 
 	// Terminate the Client.
-	defer cli.Terminate()
+	defer lib.MqttClient.Terminate()
 
 	c := client.ConnectOptions{
 		Network:  "tcp",
@@ -53,13 +49,13 @@ func Run_MqttServer() {
 	logs.Println("ClientID：", string(c.ClientID))
 
 	// Connect to the MQTT Server.
-	err := cli.Connect(&c)
+	err := lib.MqttClient.Connect(&c)
 
 	if err != nil {
 		logs.Println("MqttServer", "连接MQTT失败 [cli.Connect]", "")
 		logs.Println("err!!!!!! Run_MqttServer:", "连接MQTT失败：", err)
 		fmt.Println("err!!!!!! 连接MQTT失败:", err)
-		cli.Terminate()
+		lib.MqttClient.Terminate()
 		time.Sleep(3 * time.Second)
 		go Run_MqttServer() // MQTT 通讯
 
@@ -67,7 +63,7 @@ func Run_MqttServer() {
 	}
 	logs.Println("MqttServer", "连接成功！")
 	// Subscribe to topics.
-	err = cli.Subscribe(&client.SubscribeOptions{
+	err = lib.MqttClient.Subscribe(&client.SubscribeOptions{
 		SubReqs: []*client.SubReq{
 			&client.SubReq{
 				TopicFilter: []byte("#"),
@@ -271,145 +267,32 @@ func messagePubHandler(topicName string, message []byte) {
 	}
 
 	if Device_r.T_state != 1 {
-		logs.Println("MqttServer", Device_r.T_sn+" 设备未激活")
+		logs.Println("MqttServer", Device_r.T_sn+" 设备被 禁用、删除、无效")
 		return
 	}
 
-	// 设备类型
-	ProductType_r := Product.ProductType{T_ProductID: Device_r.T_ProductID}
-	if !ProductType_r.Read() {
-		logs.Println("MqttServer", Device_r.T_sn+"|"+Device_r.T_ProductID+" 设备类型找不到！")
-		return
-	}
-
-	// 设备协议
-	ProductProt_r := Product.ProductProt{Id: ProductType_r.T_prot}
-	if !ProductProt_r.Read() {
-		logs.Println("MqttServer", Device_r.T_sn+"|"+Device_r.T_ProductID+"-"+fmt.Sprintf("%d", ProductType_r.T_prot)+" 设备协议找不到！")
-		return
-	}
-
-	// Plugin 指针添加到 MAP，不知道能不能提高速度
-	//_, exists := lib.TopicMap[string(topicName)]
-	//if exists {
-	//	//logs.Println("跳过：",string(topicName))
-	//	return // 这个 订阅号 平台->设备，跳过
-	//}
-	//p, p_is := lib.PluginMap[ProductProt_r.T_analysis]
-	//if !p_is {
-	//	var err error
-	//	// 根据库的存放路径加载库
-	//	p_p, err := plugin.Open(conf.Analysis_Dir + ProductProt_r.T_analysis + ".so")
-	//	if err != nil {
-	//		logs.PrintlnError("打开 SO 失败：", err)
-	//		return
-	//	}
-	//	p = p_p
-	//	lib.PluginMap[ProductProt_r.T_analysis] = p
-	//	logs.Println("NEW Plugin 地址：",&p_p)
-	//}
-
-	// 加载 SO 文件
-	p, err := plugin.Open(conf.Analysis_Dir + ProductProt_r.T_analysis + ".so")
-	if err != nil {
-		logs.PrintlnError("打开 SO 失败：", err)
-		return
-	}
-	logs.Println("Plugin 地址：", &p)
-
-	var Rt_r = lib.Rt{Status: 200, Msg: "ok"}
-	// 查找库导出信息
-	s, err := p.Lookup("T")
-	if err != nil {
-		logs.Println("", err)
-		panic(any(err))
-	}
-	// 类型转换
-	f := s.(func(t string, b []byte) string)(topicName, message)
-
-	// 开始处理
-	logs.Println("协议后：", f)
-	//logs.Println("首字符：", string(f[0]))
-	if f[0] == '{' {
-		//结构体
-		var json_r map[string]interface{}
-		err = json.Unmarshal([]byte(f), &json_r)
-		if err != nil {
-			Rt_r.Status = 203
-			Rt_r.Msg = "json E!"
-			goto Mreturn
-		}
-
-		Handle.AnalysisMap(&Device_r, ProductType_r, json_r, "")
-
-		// 合并json
-		Device_r.Read_Tidy() // 提前最新的
-		logs.Println("Device_r.T_data:", Device_r.T_data)
-		if len(Device_r.T_data) > 5 {
-			json.Unmarshal([]byte(Device_r.T_data), &json_r)
-		}
-		Device_r.T_dataJson = json_r
-		Device_r.UpdateTime.NowDbTime()
-		Device_r.Update("T_data", "UpdateTime")
-
-	} else {
-		//列表
-		var json_lr []map[string]interface{}
-		err = json.Unmarshal([]byte(f), &json_lr)
-		if err != nil {
-			Rt_r.Status = 204
-			Rt_r.Msg = "[]json E!"
-			goto Mreturn
-		}
-		for _, value := range json_lr {
-			Handle.AnalysisMap(&Device_r, ProductType_r, value, "")
-		}
-
-		// 合并json
-		Device_r.Read_Tidy() // 提前最新的
-		if len(Device_r.T_data) > 5 {
-			json.Unmarshal([]byte(Device_r.T_data), &json_lr[len(json_lr)-1])
-		}
-		Device_r.T_dataJson = json_lr[len(json_lr)-1]
-		Device_r.UpdateTime.NowDbTime()
-		Device_r.Update("T_data", "UpdateTime")
-	}
+	Rt_r := Handle.PullHandle(&Device_r, topicName, message)
 
 	// 返回
-Mreturn:
 	data, _ := json.Marshal(Rt_r)
-	fmt.Println(string(data))
-
-	// 查找库导出信息
-	s, err = p.Lookup("R")
-	if err != nil {
-		logs.Println("", err)
-		panic(any(err))
-	}
-	// 转换
-	t, b := s.(func(t string, b string) (string, []byte))(topicName, string(data))
-
-	// 订阅号 与 内容 必须有数据
-	if len(t) > 0 && len(b) > 0 {
-		// 返回数据
-		Mqtt_publish(t, b)
-	}
+	Handle.PushHandle(&Device_r, topicName, string(data))
 
 }
 
-// 发送数据
-func Mqtt_publish(topic string, b []byte) {
-	lib.TopicMap[topic] = true
-	// Publish a message.
-	err := cli.Publish(&client.PublishOptions{
-		QoS:       mqtt.QoS0,
-		TopicName: []byte(topic),
-		Message:   b,
-	})
-
-	logs.PrintlnMqtt("-> " + topic + ":" + string(b))
-	if err != nil {
-		logs.PrintlnError("MqttServer", "发送消息失败 [Mqtt_publish]", "-> "+topic+" "+string(b))
-	}
-
-}
+//
+//// 发送数据
+//func Mqtt_publish(topic string, b []byte) {
+//	lib.TopicMap[topic] = true
+//	// Publish a message.
+//	err := lib.MqttClient.Publish(&client.PublishOptions{
+//		QoS:       mqtt.QoS0,
+//		TopicName: []byte(topic),
+//		Message:   b,
+//	})
+//
+//	logs.PrintlnMqtt("-> " + topic + ":" + string(b))
+//	if err != nil {
+//		logs.PrintlnError("MqttServer", "发送消息失败 [Mqtt_publish]", "-> "+topic+" "+string(b))
+//	}
+//
+//}

Handle/Relay.go

@@ -1,31 +1,84 @@
 package Handle
 
 import (
+	"Yunlot/lib"
+	"Yunlot/logs"
 	"Yunlot/models/Device"
-	"Yunlot/models/Product"
 	"encoding/json"
-	"fmt"
+	"go.mongodb.org/mongo-driver/bson"
+	"io/ioutil"
+	"net/http"
 	"strings"
+	"time"
 )
 
-func Relay(Device_r Device.Device, ProductType_r Product.ProductType, JointTab string, ArticleSlide map[string]interface{}) {
-	ProductRelay_r := Product.ProductRelay{T_ProductID: Device_r.T_ProductID,T_rtab: JointTab}
-	if ProductRelay_r.Read() {
-		datajson, _ := json.Marshal(ArticleSlide)
-		fmt.Println(string(datajson))
-
-		//  替换发布号变量
-		ProductRelay_r.T_pub = strings.Replace(ProductRelay_r.T_pub, "{sn}", Device_r.T_sn, -1)
-		if len(ProductRelay_r.T_pub) == 0 {
-			println("订阅号错误！")
-			return
-		}
-		switch ProductRelay_r.T_mode {
-		case 0:
+func Relay(Device_r *Device.Device, JointTab string, ArticleSlide *bson.M) {
+	for key, value := range Device_r.T_ProductJson.T_RelayDataJson {
+		if key == JointTab || key == "*" {
+			x := value.(map[string]interface{})
+
+			//  替换发布号变量
+			T_pub := strings.Replace(lib.To_string(x["T_pub"]), "{$sn}", Device_r.T_sn, -1)
+			if len(T_pub) == 0 {
+				logs.Println("订阅号错误！")
+				return
+			}
+			// 将bson.M对象转换为JSON格式
+			jsonData, err := json.Marshal(ArticleSlide)
+			if err != nil {
+				continue
+			}
+			switch lib.To_int(x["T_mode"]) {
+			case 0: //  Nats
+				logs.Println("Relay -> Nats:", T_pub, string(jsonData))
+				err = lib.Nats.Publish("T_pub", jsonData)
+				if err != nil {
+					// 发送失败3秒后再次尝试
+					go func() {
+						time.Sleep(time.Second * 3)
+						lib.Nats.Publish("T_pub", jsonData)
+					}()
+				}
+				break
+			case 1: //  API
+				logs.Println("Relay -> API:", T_pub, string(jsonData))
+
+				go func() {
+					payload := strings.NewReader(string(jsonData))
 
-			break
+					client := &http.Client{
+						Timeout: 3 * time.Second, // 设置超时时间为5秒
+					}
+					req, err := http.NewRequest("POST", T_pub, payload)
+					if err != nil {
+						logs.Println("Relay -> API!!!:", T_pub, any(err))
+						return
+					}
+					req.Header.Add("Content-Type", "application/json")
 
+					res, err := client.Do(req)
+					if err != nil {
+						logs.Println("Relay -> API!!!:", T_pub, any(err))
+						return
+					}
+					defer res.Body.Close()
+
+					body, err := ioutil.ReadAll(res.Body)
+					if err != nil {
+						logs.Println("Relay -> API!!!:", T_pub, any(err))
+						return
+					}
+					logs.Println("Relay -> API Body:", string(body))
+				}()
+
+				break
+			case 2: //  MQTT
+				logs.Println("Relay -> MQTT:", T_pub, string(jsonData))
+				lib.Mqtt_publish(T_pub, jsonData)
+				break
+			}
 		}
 
 	}
+
 }

RunCode/config/main.txt

@@ -15,7 +15,7 @@ func main() {
 	}
 	// 平台->设备（接收）：
 	if len(R_string) != 0{
-		topic_, data_ := R(Topic, R_string)
+		topic_, data_ := R("{$sn}", R_string)
 		fmt.Print("|-=&" + topic_)
 		if Data_hex {
 			fmt.Print("|-=&" + strings.ToUpper(hex.EncodeToString(data_)))

RunCode/config/main.txt_test.go

@@ -4,7 +4,6 @@ import (
 	"testing"
 )
 
-
 /////-------
 import (
 	"encoding/hex"
@@ -15,16 +14,16 @@ import (
 /*
 设备->平台
 */
-func T(t string, b []byte) (string) {
+func T(t string, b []byte) string {
 	return string(b)
 }
 
-
 /*
 平台->设备
 */
-func R(t string, b string) (string,[]byte){
-	return t,[]byte(b)
+func R(sn string, b string) (string, []byte) {
+	t := sn
+	return t, []byte(b)
 }
 
 var Data_hex = true
@@ -32,7 +31,6 @@ var Topic = "/topic/snxxx1"
 var T_string = "A456"
 var R_string = "495051"
 
-
 /////-------
 
 func main() {
@@ -43,30 +41,29 @@ func main() {
 		T_byte, _ = hex.DecodeString(T_string)
 	}
 	// 设备->平台（发送）
-	if len(T_string) != 0{
+	if len(T_string) != 0 {
 		data_ := T(Topic, T_byte)
 		fmt.Print("|-=&" + data_)
-	}else{
+	} else {
 		fmt.Print("|-=&" + "无参数")
 	}
 	// 平台->设备（接收）：
-	if len(R_string) != 0{
+	if len(R_string) != 0 {
 		topic_, data_ := R(Topic, R_string)
 		fmt.Print("|-=&" + topic_)
 		if Data_hex {
 			fmt.Print("|-=&" + strings.ToUpper(hex.EncodeToString(data_)))
-		}else {
+		} else {
 			fmt.Print("|-=&" + string(data_))
 		}
 
-	}else{
+	} else {
 		fmt.Print("|-=&" + "无参数")
 		fmt.Print("|-=&" + "无参数")
 	}
 	// 返回数据： 1、 设备发送数据解析到平台json     2、topic     3、平台处理后返回给设备数据
 }
 
-
 func Test_Runcode(t *testing.T) {
 	main()
 }

controllers/Device.go

@@ -1,6 +1,7 @@
 package controllers
 
 import (
+	"Yunlot/Handle"
 	"Yunlot/lib"
 	"Yunlot/models/Device"
 	"Yunlot/models/Product"
@@ -76,10 +77,11 @@ func (c *DeviceController) Add() {
 }
 
 func (c *DeviceController) Update() {
-	Devicer := Device.Device{}
+	Devicer := Device.Device{T_sn: c.GetString("T_sn")}
+	Devicer.Read_Tidy()
 	c.ParseForm(&Devicer)
 
-	if !Devicer.Update("T_name", "T_img", "T_prot") {
+	if !Devicer.Update("T_state") {
 		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "E!"}
 		c.ServeJSON()
 		return
@@ -119,6 +121,27 @@ func (c *DeviceController) Get() {
 	c.ServeJSON()
 	return
 }
+func (c *DeviceController) Push() {
+	Devicer := Device.Device{}
+	c.ParseForm(&Devicer)
+	T_data := c.GetString("T_data")
+	if len(T_data) == 0 {
+		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "T_data E!"}
+		c.ServeJSON()
+		return
+	}
+	if !Devicer.Read_Tidy() {
+		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "SN E!"}
+		c.ServeJSON()
+		return
+	}
+
+	Handle.PushHandle(&Devicer, Devicer.T_sn, T_data)
+
+	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!"}
+	c.ServeJSON()
+	return
+}
 
 func (c *DeviceController) DataList() {
 

controllers/ProductRelay.go

@@ -1,68 +0,0 @@
-package controllers
-
-import (
-	"Yunlot/lib"
-	"Yunlot/models/Product"
-	beego "github.com/beego/beego/v2/server/web"
-)
-
-type ProductRelayController struct {
-	beego.Controller
-}
-
-func (c *ProductRelayController) List() {
-	PageIndex, _ := c.GetInt("PageIndex", 0)
-	PageSize, _ := c.GetInt("PageSize", 10)
-
-	ProductRelayr := Product.ProductRelay{}
-	c.ParseForm(&ProductRelayr)
-
-	ProductRelay_r, Total := ProductRelayr.Lists(PageIndex, PageSize)
-
-	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!", Data: lib.C_Page(ProductRelay_r, PageIndex, PageSize, Total)}
-	c.ServeJSON()
-	return
-}
-
-func (c *ProductRelayController) Add() {
-	ProductRelay_r := Product.ProductRelay{}
-	c.ParseForm(&ProductRelay_r)
-	ProductRelay_r.Id = 0
-	ProductRelay_r.Add()
-
-	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!", Data: ProductRelay_r}
-	c.ServeJSON()
-	return
-}
-
-func (c *ProductRelayController) Update() {
-	ProductRelayr := Product.ProductRelay{}
-	c.ParseForm(&ProductRelayr)
-
-	if !ProductRelayr.Update("T_name", "T_rtab", "T_mode", "T_pub") {
-		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "E!"}
-		c.ServeJSON()
-		return
-	}
-
-	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!"}
-	c.ServeJSON()
-	return
-}
-
-func (c *ProductRelayController) Delete() {
-	ProductRelayr := Product.ProductRelay{}
-	c.ParseForm(&ProductRelayr)
-
-	if !ProductRelayr.Delete() {
-		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "E!"}
-		c.ServeJSON()
-		return
-	}
-
-	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!"}
-	c.ServeJSON()
-	return
-}
-
-

controllers/ProductType.go

@@ -2,6 +2,7 @@ package controllers
 
 import (
 	"Yunlot/lib"
+	"Yunlot/models/Device"
 	"Yunlot/models/Product"
 	beego "github.com/beego/beego/v2/server/web"
 )
@@ -54,15 +55,23 @@ func (c *ProductTypeController) Add() {
 }
 
 func (c *ProductTypeController) Update() {
-	ProductTyper := Product.ProductType{}
+
+	ProductTyper := Product.ProductType{T_ProductID: c.GetString("T_ProductID")}
+	if !ProductTyper.Read() {
+		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "T_ProductID E!"}
+		c.ServeJSON()
+		return
+	}
+
 	c.ParseForm(&ProductTyper)
 
-	if !ProductTyper.Update("T_name", "T_img", "T_prot", "T_TabData") {
+	if !ProductTyper.Update("T_name", "T_img", "T_prot", "T_TabData", "T_RelayData") {
 		c.Data["json"] = lib.JSONR{Code: lib.Error, Msg: "E!"}
 		c.ServeJSON()
 		return
 	}
 
+	Device.Device_CacheDelK_All(ProductTyper.T_ProductID) // 删除 关联设备缓存
 	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!"}
 	c.ServeJSON()
 	return
@@ -77,7 +86,7 @@ func (c *ProductTypeController) Delete() {
 		c.ServeJSON()
 		return
 	}
-
+	Device.Device_CacheDelK_All(ProductTyper.T_ProductID) // 删除 关联设备缓存
 	c.Data["json"] = lib.JSONR{Code: lib.Success, Msg: "ok!"}
 	c.ServeJSON()
 	return

lib/Struct.go

@@ -1,8 +0,0 @@
-package lib
-
-// /  -   发送 实体
-type Rt struct {
-	//Msid   int `json:"msid"`
-	Status int    `json:"Status"` // 200
-	Msg    string `json:"Msg"`
-}

lib/lib.go

@@ -1,6 +1,7 @@
 package lib
 
 import (
+	"Yunlot/logs"
 	"bytes"
 	"crypto/sha256"
 	"encoding/hex"
@@ -8,11 +9,14 @@ import (
 	"fmt"
 	"github.com/nats-io/nats.go"
 	"github.com/thinkeridea/go-extend/exunicode/exutf8"
+	"github.com/yosssi/gmq/mqtt"
+	"github.com/yosssi/gmq/mqtt/client"
 	"golang.org/x/text/encoding/simplifiedchinese"
 	"golang.org/x/text/transform"
 	"io/ioutil"
 	"math/rand"
 	"plugin"
+	"reflect"
 	"runtime"
 	"strconv"
 	"strings"
@@ -21,6 +25,7 @@ import (
 )
 
 var Nats *nats.Conn
+var MqttClient *client.Client
 var TopicMap map[string]bool
 var PluginMap map[string]*plugin.Plugin
 
@@ -436,3 +441,39 @@ func WeekByDate() string {
 	}
 	return fmt.Sprintf("%d%02d", t.Year(), week) // 202253
 }
+
+// 发送数据
+func Mqtt_publish(topic string, b []byte) {
+	TopicMap[topic] = true
+	// Publish a message.
+	err := MqttClient.Publish(&client.PublishOptions{
+		QoS:       mqtt.QoS0,
+		TopicName: []byte(topic),
+		Message:   b,
+	})
+
+	logs.PrintlnMqtt("Mqtt-> " + topic + ":" + string(b))
+	if err != nil {
+		logs.PrintlnError("MqttServer", "发送消息失败 [Mqtt_publish]", "-> "+topic+" "+string(b))
+	}
+
+}
+
+// 获取json指定参数   [AAAA BBBB],0,JSON    ！！！如果失败，返回已找到的全部json
+func Json_key(topicNameList []string, topicNameI int, articleSlide map[string]interface{}) interface{} {
+	if len(topicNameList) > topicNameI {
+		a, is := articleSlide[topicNameList[topicNameI]]
+		if !is {
+			return articleSlide
+		}
+		if reflect.TypeOf(a).String() == "map[string]interface {}" {
+			return Json_key(topicNameList, topicNameI+1, a.(map[string]interface{}))
+		}
+	}
+
+	if len(topicNameList) == topicNameI {
+		return articleSlide
+	}
+
+	return articleSlide[topicNameList[topicNameI]]
+}

main.go

@@ -11,6 +11,7 @@ import (
 	"Yunlot/logs"
 	"Yunlot/models/Product"
 	_ "Yunlot/routers"
+	"fmt"
 	"github.com/beego/beego/v2/adapter/orm"
 	orm2 "github.com/beego/beego/v2/client/orm"
 	beego "github.com/beego/beego/v2/server/web"
@@ -46,7 +47,7 @@ func main() {
 		mode := Product.ProductMode{
 			Id:         1,
 			T_name:     "MQTT",
-			T_address:  "mqtt://bj-3-mqtt.iot-api.com:1883",
+			T_address:  fmt.Sprintf("mqtt://%s", conf.MqttServer_Url),
 			T_describe: "MQTT 接入方式为设备和云平台提供双向连接，设备既可上报属性数据，也可接收云端的消息下发。",
 			T_state:    1,
 		}
@@ -70,7 +71,7 @@ func main() {
 		mode := Product.ProductMode{
 			Id:         6,
 			T_name:     "HTTP",
-			T_address:  "HTTP://bj-3-mqtt.iot-api.com:1883",
+			T_address:  fmt.Sprintf("http://:%d", conf.HTTPServer_Port),
 			T_describe: "HTTP 接入方式为设备和云平台提供双向连接，设备既可上报属性数据，也可接收云端的消息下发。",
 			T_state:    1,
 		}

models/Device/Device.go

@@ -20,15 +20,15 @@ import (
 type Device struct {
 	T_sn          string                 `orm:"size(256);pk" json:"T_sn" form:"T_sn"`                     // Sn
 	T_password    string                 `orm:"size(256);" json:"T_password" form:"T_password"`           // 密码
-	T_online      int                    `orm:"size(1);index;default(0)" json:"T_online" form:"T_online"` // 在线状态  0 离线  1 在线   3 无效
+	T_online      int                    `orm:"size(1);index;default(3)" json:"T_online" form:"T_online"` // 在线状态  1 在线  2 离线   3 未激活
 	T_ProductID   string                 `orm:"size(8);index" json:"T_ProductID" form:"T_ProductID"`      // 产品类型
 	T_ProductJson Product.ProductType    `orm:"-" json:"T_ProductJson"`                                   // 产品类型 json
 	T_data        string                 `orm:"column(t_data);type(text);default('')" json:"T_data" `     // 设备数据
 	T_dataJson    map[string]interface{} `orm:"-" json:"T_dataJson"`                                      // 设备数据
 
-	T_state    int         `orm:"size(2);default(0);index" json:"T_state" form:"T_state"` // 0 未激活   1 正常  2 禁用\删除
-	CreateTime models.Time `orm:"column(create_time);type(timestamp);auto_now_add"`
-	UpdateTime models.Time `orm:"column(update_time);type(timestamp);auto_now"`
+	T_state    int         `orm:"size(2);default(1);index" json:"T_state" form:"T_state"` //  1 正常  2 禁用\删除   3 无效
+	CreateTime models.Time `orm:"column(create_time);type(timestamp);auto_now_add" json:"CreateTime"`
+	UpdateTime models.Time `orm:"column(update_time);type(timestamp);auto_now" json:"UpdateTime"`
 }
 
 func (t *Device) TableName() string {
@@ -52,8 +52,7 @@ func init() {
 }
 
 // ---------------- Redis -------------------
-func (t *Device) redis_Set() (err error) {
-
+func (t *Device) Redis_Set() (err error) {
 	if len(t.T_data) > 5 {
 		var T_datajson map[string]interface{}
 		json.Unmarshal([]byte(t.T_data), &T_datajson)
@@ -77,31 +76,33 @@ func (t *Device) redis_Set() (err error) {
 	}
 	return
 }
-func (t *Device) redis_Get(key string) (is bool) {
+func (t *Device) Redis_Get(key string) (is bool) {
 	if redis_Device.IsExist(key) {
 		//println("找到key:",key)
 		v := redis_Device.Get(key)
 		if v == nil {
 			return false
 		}
+
 		json.Unmarshal(v.([]byte), t)
 
 		if t.T_state == 3 { //  3 无效
 			return false
 		}
+
 		return true
 	}
 	//println("没有 找到key:",key)
 	return false
 }
-func (t *Device) redis_DelK() (err error) {
+func (t *Device) Redis_DelK() (err error) {
 	err = redis_Device.Delete(t.T_sn)
 	return
 }
 
 // ---------------- 方法 -------------------
 
-// 获取
+// 获取(不带缓存查询)
 func (t *Device) Read() (is bool) {
 	o := orm.NewOrm()
 	err := o.Read(t) // o.Read(&r,"Tokey") 如果不是 主键 就得指定字段名
@@ -109,23 +110,23 @@ func (t *Device) Read() (is bool) {
 		return false
 	}
 
-	t.redis_Set()
+	t.Redis_Set()
 	return true
 }
 
 // 获取-高效的
 func (t *Device) Read_Tidy() (bool bool) {
-	if t.redis_Get(t.T_sn) {
+	if t.Redis_Get(t.T_sn) {
 		return true
 	}
 	o := orm.NewOrm()
 	err := o.Read(t) // o.Read(&r,"Tokey") 如果不是 主键 就得指定字段名
 	if err != nil {
-		t.T_state = 3 // 3 无效
-		t.redis_Set() // Redis 更新缓存
+		//t.T_state = 3 // 3 无效
+		//t.Redis_Set() // Redis 更新缓存
 		return false
 	}
-
+	t.Redis_Set() // Redis 更新缓存
 	return true
 }
 
@@ -170,7 +171,7 @@ func (t *Device) Add() (is bool) {
 		return false
 	}
 	println(id)
-	t.redis_Set()
+	t.Redis_Set()
 	return true
 }
 
@@ -182,7 +183,7 @@ func (t *Device) Update(cols ...string) bool {
 	o := orm.NewOrm()
 	if num, err := o.Update(t, cols...); err == nil {
 		logs.Println("Number of records updated in database:", num)
-		t.redis_Set() // Redis 更新缓存
+		t.Redis_Set() // Redis 更新缓存
 		return true
 	}
 	return false
@@ -197,7 +198,7 @@ func (t *Device) Delete() bool {
 		return false
 	}
 
-	t.redis_DelK()
+	t.Redis_DelK()
 	return true
 }
 
@@ -220,6 +221,12 @@ func (t *Device) Lists(PageIndex int, PageSize int) (r []Device, Total int64) {
 	if len(t.T_ProductID) > 0 {
 		cond = cond.And("T_ProductID", t.T_ProductID) // .AndNot("status__in", 1).Or("profile__age__gt", 2000)
 	}
+	if len(t.T_sn) > 0 {
+		cond = cond.And("T_sn__icontains", t.T_sn) // .AndNot("status__in", 1).Or("profile__age__gt", 2000)
+	}
+	if t.T_online > 0 {
+		cond = cond.And("T_online", t.T_online) // .AndNot("status__in", 1).Or("profile__age__gt", 2000)
+	}
 
 	// 执行
 	qs.Limit(PageSize, offset).SetCond((*orm2.Condition)(cond)).All(&r)
@@ -267,3 +274,24 @@ func (t *Device) Lists_All() (r []Device) {
 	}
 	return r
 }
+
+// 删除 关联设备缓存
+func Device_CacheDelK_All(T_ProductID string) {
+
+	o := orm.NewOrm()
+
+	// 也可以直接使用 Model 结构体作为表名
+	qs := o.QueryTable(new(Device))
+
+	// 筛选参数
+	cond := orm.NewCondition()
+	cond = cond.And("T_ProductID", T_ProductID) // .AndNot("status__in", 1).Or("profile__age__gt", 2000)
+
+	// 执行
+	var r []Device
+	qs.SetCond((*orm2.Condition)(cond)).All(&r)
+	for _, v := range r {
+		v.Redis_DelK()
+	}
+	return
+}

models/Product/ProductProt.go

@@ -19,7 +19,7 @@ type ProductProt struct {
 	T_name string `orm:"size(256);" json:"T_name" form:"T_name"`         // 协议名称
 	T_mode int    `orm:"size(1);default(0)" json:"T_Mode" form:"T_mode"` //接入方式  0：Mqtt  1：http  2:tcp  3:CoAP  4:websocket
 	//T_prot     int    `orm:"size(1);default(0)" json:"T_Prot" form:"T_prot"`                              //接入协议ID  0：统一协议
-	T_lang     int    `orm:"size(1);default(1)" json:"T_lang" form:"T_lang"`                                  //编程语言  0: C  1: go
+	T_lang     int    `orm:"size(1);default(1)" json:"T_lang" form:"T_lang"`                                  //编程语言 0: 无解析 1: go   2: C
 	T_analysis string `orm:"size(100);default(yunlot-pxxxx-v1.19-01.so)" json:"T_analysis" form:"T_analysis"` //数据解析
 	T_text     string `orm:"type(text);default('')" json:"T_text" form:"T_text"`                              // 代码内容
 	T_describe string `orm:"type(text);default('')" json:"T_describe" form:"T_describe"`                      // 描述内容

models/Product/ProductRelay.go

@@ -1,159 +0,0 @@
-package Product
-
-import (
-	"Yunlot/conf"
-	"Yunlot/logs"
-	"encoding/json"
-	"fmt"
-	"github.com/astaxie/beego/cache"
-	_ "github.com/astaxie/beego/cache/redis"
-	"github.com/beego/beego/v2/adapter/orm"
-	orm2 "github.com/beego/beego/v2/client/orm"
-	_ "github.com/go-sql-driver/mysql"
-	"time"
-)
-
-// 消息转发
-type ProductRelay struct {
-	Id            int    `orm:"column(ID);size(11);auto;pk" json:"ID"`
-	T_ProductID string `orm:"size(8);index" json:"T_ProductID" form:"T_ProductID"` // 产品型号 随机生成（8位）
-
-	T_name string `orm:"size(256);" json:"T_name" form:"T_name"`         // 名称
-	T_rtab string `orm:"size(32);index" json:"T_rtab" form:"T_rtab"`     //转发 TAB 标识 拼接：AAAA.BBBB.   对象后面加点
-	T_mode int    `orm:"size(1);default(0)" json:"T_mode" form:"T_mode"` //接入方式  0：Mqtt  1：http  2:tcp  3:CoAP  4:websocket
-	T_pub  string `orm:"size(256);" json:"T_pub" form:"T_pub"`           //发布号
-
-}
-
-func (t *ProductRelay) TableName() string {
-	return "ProductRelay" // 数据库名称   // ************** 替换 FormulaList **************
-}
-
-var redis_ProductRelay cache.Cache
-
-func init() {
-	//注册模型
-	orm.RegisterModel(new(ProductRelay))
-
-	config := fmt.Sprintf(`{"key":"%s","conn":"%s","dbNum":"%s","password":"%s"}`,
-		"redis_ProductRelay", conf.Redis_address, conf.Redis_dbNum, conf.Redis_password)
-	var err error
-	redis_ProductRelay, err = cache.NewCache("redis", config)
-	if err != nil || redis_ProductRelay == nil {
-		logs.Println(config)
-		panic(any(err))
-	}
-}
-
-// ---------------- Redis -------------------
-func (t *ProductRelay) redis_Set() (err error) {
-	//json序列化
-	str, err := json.Marshal(t)
-	if err != nil {
-		logs.PrintlnError("Redis_Set", err)
-		return
-	}
-
-	err = redis_ProductRelay.Put(t.T_ProductID+"-"+t.T_rtab, str, 24*time.Hour)
-	if err != nil {
-		logs.Println("set key:", t.T_ProductID+"-"+t.T_rtab, ",value:", str, err)
-	}
-	return
-}
-func (t *ProductRelay) redis_Get(key string) (is bool) {
-	if redis_ProductRelay.IsExist(key) {
-		//println("找到key:",key)
-		v := redis_ProductRelay.Get(key)
-		if v == nil {
-			return false
-		}
-		json.Unmarshal(v.([]byte), t)
-		return true
-	}
-	//println("没有 找到key:",key)
-	return false
-}
-func (t *ProductRelay) redis_DelK() (err error) {
-	err = redis_ProductRelay.Delete(t.T_ProductID)
-	return
-}
-
-// ---------------- 方法 -------------------
-
-// 添加
-func (t *ProductRelay) Add() (is bool) {
-	o := orm.NewOrm()
-
-	id, err := o.Insert(t)
-	if err != nil {
-		return false
-	}
-	println(id)
-	t.redis_Set()
-	return true
-}
-
-// 获取
-func (t *ProductRelay) Read() (bool bool) {
-	if t.redis_Get(t.T_ProductID+"-"+t.T_rtab) {
-		return true
-	}
-	o := orm.NewOrm()
-	err := o.Read(&t) // o.Read(&r,"Tokey") 如果不是 主键 就得指定字段名
-	if err != nil {
-		return false
-	}
-	t.redis_Set() // Redis 更新缓存
-	return true
-}
-
-// 修改
-func (t *ProductRelay) Update(cols ...string) bool {
-	o := orm.NewOrm()
-	if num, err := o.Update(t, cols...); err == nil {
-		logs.Println("Number of records updated in database:", num)
-		t.redis_Set() // Redis 更新缓存
-		return true
-	}
-	return false
-}
-
-// 删除
-func (t *ProductRelay) Delete() bool {
-	o := orm.NewOrm()
-	if num, err := o.Delete(t); err == nil {
-		logs.Println("Number of records deleted in database:", num)
-	} else {
-		return false
-	}
-
-	t.redis_DelK()
-	return true
-}
-
-// 获取列表
-func (t *ProductRelay) Lists(PageIndex int, PageSize int) (r []ProductRelay, Total int64) {
-
-	o := orm.NewOrm()
-
-	// 也可以直接使用 Model 结构体作为表名
-	qs := o.QueryTable(new(ProductRelay))
-	var offset int64
-	if PageIndex <= 1 {
-		offset = 0
-	} else {
-		offset = int64((PageIndex - 1) * PageSize)
-	}
-
-	// 筛选参数
-	cond := orm.NewCondition()
-	if len(t.T_ProductID) > 0 {
-		cond = cond.And("T_ProductID", t.T_ProductID) // .AndNot("status__in", 1).Or("profile__age__gt", 2000)
-	}
-
-	// 执行
-	qs.Limit(PageSize, offset).SetCond((*orm2.Condition)(cond)).All(&r)
-	Total, _ = qs.SetCond((*orm2.Condition)(cond)).Count()
-
-	return r, Total
-}

models/Product/ProductType.go

@@ -22,18 +22,18 @@ type ProductType struct {
 	T_akey      string `orm:"size(56);" json:"T_akey" `                         // 授权密钥
 	T_prot      int    `orm:"size(10);default(0)" json:"T_prot" form:"T_prot"`  //接入协议ID  0：统一协议
 
-	T_TabData     string                   `orm:"type(text);default('')" json:"T_TabData" form:"T_TabData"` // 产品模型
-	T_TabDataJson []map[string]interface{} `orm:"-" json:"T_TabDataJson"`                                   // 产品模型json
+	T_TabData     string                   `orm:"type(text);default('{}')" json:"T_TabData" form:"T_TabData"` // 产品模型
+	T_TabDataJson []map[string]interface{} `orm:"-" json:"T_TabDataJson"`                                     // 产品模型 json
 
-	CreateTime models.Time `orm:"column(create_time);type(timestamp);auto_now_add"`
-	UpdateTime models.Time `orm:"column(update_time);type(timestamp);auto_now"`
-}
+	T_RelayData     string                 `orm:"type(text);default('{}')" json:"T_RelayData" form:"T_RelayData"` // 消息转发
+	T_RelayDataJson map[string]interface{} `orm:"-" json:"T_RelayDataJson"`                                       // 消息转发 json
 
-/*
-	T_TabData、T_TabDataJson
+	CreateTime models.Time `orm:"column(create_time);type(timestamp);auto_now_add" json:"CreateTime"`
+	UpdateTime models.Time `orm:"column(update_time);type(timestamp);auto_now" json:"UpdateTime"`
+}
 
+/*  产品模型  T_TabData \ T_TabDataJson
 [
-
 	{
 		"T_name": "温度",
 		"T_tab": "Tt",
@@ -83,6 +83,28 @@ type ProductType struct {
 			value 值
 			text 描述
 */
+
+/*
+	 消息转发  T_RelayData \ T_RelayDataJson
+		{
+			"AAAA.BBBB": {
+				"T_name": "BBBB转发",
+				"T_mode": 0,
+				"T_pub": "aaa/{$sn}/bbbb"
+			},
+			"AAAA.CCCC": {
+				"T_name": "CCCC转发",
+				"T_mode": 2,
+				"T_pub": "aaa/{$sn}/cccc"
+			}
+		}
+
+-----------------
+//  {KEY}   监听 TAB 标识 拼接：AAAA.BBBB   对象后面加点
+//  	T_name   名称
+//  	T_mode   推送服务 0：Nats  1：API  2:MQTT
+//  	T_pub   发布号 /aaa/{$sn}/bbb
+*/
 func (t *ProductType) TableName() string {
 	return "ProductType" // 数据库名称   // ************** 替换 FormulaList **************
 }
@@ -111,6 +133,11 @@ func (t *ProductType) redis_Set() (err error) {
 		json.Unmarshal([]byte(t.T_TabData), &T_datajson)
 		t.T_TabDataJson = T_datajson
 	}
+	if len(t.T_RelayData) > 0 {
+		var T_datajson map[string]interface{}
+		json.Unmarshal([]byte(t.T_RelayData), &T_datajson)
+		t.T_RelayDataJson = T_datajson
+	}
 
 	//json序列化
 	str, err := json.Marshal(t)
@@ -138,7 +165,7 @@ func (t *ProductType) redis_Get(key string) (is bool) {
 	//println("没有 找到key:",key)
 	return false
 }
-func (t *ProductType) redis_DelK() (err error) {
+func (t *ProductType) Redis_DelK() (err error) {
 	err = redis_ProductType.Delete(t.T_ProductID)
 	return
 }
@@ -179,6 +206,7 @@ func (t *ProductType) Update(cols ...string) bool {
 	if err == nil {
 		logs.Println("Number of records updated in database:", num)
 		t.redis_Set() // Redis 更新缓存
+
 		return true
 	}
 	fmt.Println(err)
@@ -194,7 +222,8 @@ func (t *ProductType) Delete() bool {
 		return false
 	}
 
-	t.redis_DelK()
+	t.Redis_DelK()
+
 	return true
 }
 
@@ -230,7 +259,13 @@ func (t *ProductType) Lists(PageIndex int, PageSize int) (r []ProductType, Total
 			var T_datajson []map[string]interface{}
 			json.Unmarshal([]byte(r[i].T_TabData), &T_datajson)
 			r[i].T_TabDataJson = T_datajson
-			logs.Println("T_datajson", r[i].T_TabDataJson)
+			//logs.Println("T_datajson", r[i].T_TabDataJson)
+		}
+		if len(r[i].T_RelayData) > 0 {
+			var T_datajson map[string]interface{}
+			json.Unmarshal([]byte(r[i].T_RelayData), &T_datajson)
+			r[i].T_RelayDataJson = T_datajson
+			//logs.Println("T_RelayData", r[i].T_RelayDataJson)
 		}
 
 	}

routers/Device.go

@@ -11,6 +11,7 @@ func init() {
 	beego.Router(prefix+"/Device/Create", &controllers.DeviceController{}, "*:Add") // 获取未读消息
 
 	beego.Router(prefix+"/Device/Get", &controllers.DeviceController{}, "*:Get")       // 获取未读消息
+	beego.Router(prefix+"/Device/Push", &controllers.DeviceController{}, "*:Push")     // 获取未读消息
 	beego.Router(prefix+"/Device/List", &controllers.DeviceController{}, "*:List")     // 获取未读消息
 	beego.Router(prefix+"/Device/Update", &controllers.DeviceController{}, "*:Update") // 获取未读消息
 	beego.Router(prefix+"/Device/Delete", &controllers.DeviceController{}, "*:Delete") // 获取未读消息

routers/Product.go

@@ -24,10 +24,4 @@ func init() {
 	beego.Router(prefix+"/ProductProt/Update", &controllers.ProductProtController{}, "*:Update") // 获取未读消息
 	beego.Router(prefix+"/ProductProt/Delete", &controllers.ProductProtController{}, "*:Delete") // 获取未读消息
 
-	// 消息转发
-	beego.Router(prefix+"/ProductRelay/Add", &controllers.ProductRelayController{}, "*:Add")       // 获取未读消息
-	beego.Router(prefix+"/ProductRelay/List", &controllers.ProductRelayController{}, "*:List")     // 获取未读消息
-	beego.Router(prefix+"/ProductRelay/Update", &controllers.ProductRelayController{}, "*:Update") // 获取未读消息
-	beego.Router(prefix+"/ProductRelay/Delete", &controllers.ProductRelayController{}, "*:Delete") // 获取未读消息
-
 }

tests/json_key_test.go

@@ -0,0 +1,56 @@
+package test
+
+import (
+	"Yunlot/lib"
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func Test_jsonkey(t *testing.T) {
+
+	var data = `{
+	"clientID": "869023065073597",
+	"msgID": "24152",
+	"msgID11": "1111",
+	"msgID22": "1111",
+	"msgID33": "3333",
+	"AAAA": {
+		"typeCode": "123",
+		"typeCode2": "123",
+		"AAAA": {
+			"name": "TempSet",
+			"value": "272"
+		},
+		"BBBB": [{
+			"name": "TempSet",
+			"value": "266666"
+		}],
+		"CCCC": [{
+			"name": "TempSet",
+			"value": "2766662"
+		}]
+	},
+	"paramsb": {
+		"typeCode": "123",
+		"typeCode2": "123"
+	},
+	"paramslist": [{
+		"typeCode": "123",
+		"typeCode2": "123"
+	}]
+}`
+	var articleSlide map[string]interface{}
+	err := json.Unmarshal([]byte(data), &articleSlide)
+	if err != nil {
+
+	}
+	topicName := "/paramsb/"
+	topicName = strings.Trim(topicName, "/")
+
+	topicNameList := strings.Split(topicName, "/")
+
+	x := lib.Json_key(topicNameList, 0, articleSlide)
+	println(x)
+
+}

tests/logs/logx/logx.log

@@ -50,3 +50,6 @@
 2023/04/24 09:53:47.267 [I]  {"key":"redis_Company","conn":"","dbNum":"","password":""}%!(EXTRA []interface {}=[])
 2023/04/24 09:53:47.267 [I]  failed to init redis%!(EXTRA []interface {}=[dial tcp: missing address])
 2023/04/24 09:53:47.267 [I]  failed to init redis%!(EXTRA []interface {}=[dial tcp: missing address])
+2023/11/03 16:19:23.700 [I]  topicNameS:%!(EXTRA []interface {}=[[ AAAA typeCode2 ]])
+2023/11/03 16:19:33.555 [I]  topicNameS[0]:%!(EXTRA []interface {}=[])
+2023/11/03 16:20:29.371 [I]  topicNameS:%!(EXTRA []interface {}=[[ AAAA typeCode2 ]])