ipole_edge_server/edge/ipole/mqtthandle.go

package main

import (
	"os"
	"strings"

	"lc/common/mqtt"
	"lc/common/protocol"
	"lc/common/util"
)

func HandleTpQApp(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_MutilFileObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		//读文件内容
		ReadMutilFileContent(protocol.TP_GW_APP, obj.Data.Id, &ret)
		if str, err := ret.EnCode(appConfig.GID, obj.Seq); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_APP_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpWApp(m mqtt.Message) {
	var obj protocol.Pack_SeqFileObject
	var ret protocol.Pack_Ack
	err := obj.DeCode(m.PayloadString())
	if err == nil {
		err = HandleFile(protocol.TP_GW_SET_APP, &obj)
	}
	if str, err := ret.EnCode(appConfig.GID, appConfig.GID, obj.Seq, err); err == nil {
		GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_APP_ACK), str, 0, ToCloud)
	}
}
func HandleTpQSerial(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_MutilFileObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		//读文件内容
		ReadMutilFileContent(protocol.TP_GW_SERIAL, obj.Data.Id, &ret)
		if str, err := ret.EnCode(appConfig.GID, obj.Seq); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SERIAL_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpWSerial(m mqtt.Message) {
	var obj protocol.Pack_SeqFileObject
	var ret protocol.Pack_Ack
	err := obj.DeCode(m.PayloadString())
	if err == nil {
		err = HandleFile(protocol.TP_GW_SET_SERIAL, &obj)
	}
	if str, err := ret.EnCode(appConfig.GID, appConfig.GID, obj.Seq, err); err == nil {
		GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_SERIAL_ACK), str, 0, ToCloud)
	}
}
func HandleTpQRtu(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_MutilFileObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		//读文件内容
		ReadMutilFileContent(protocol.TP_GW_RTU, obj.Data.Id, &ret)
		if str, err := ret.EnCode(appConfig.GID, obj.Seq); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_RTU_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpWRtu(m mqtt.Message) {
	var obj protocol.Pack_SeqFileObject
	var ret protocol.Pack_Ack
	err := obj.DeCode(m.PayloadString())
	if err == nil {
		err = HandleFile(protocol.TP_GW_SET_RTU, &obj)
	}
	if str, err := ret.EnCode(appConfig.GID, appConfig.GID, obj.Seq, err); err == nil {
		GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_RTU_ACK), str, 0, ToCloud)
	}
}
func HandleTpQModel(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_MutilFileObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		//读文件内容
		ReadMutilFileContent(protocol.TP_GW_MODEL, obj.Data.Id, &ret)
		if str, err := ret.EnCode(appConfig.GID, obj.Seq); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_MODEL_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpWModel(m mqtt.Message) {
	var obj protocol.Pack_SeqFileObject
	var ret protocol.Pack_Ack
	err := obj.DeCode(m.PayloadString())
	if err == nil {
		err = HandleFile(protocol.TP_GW_SET_MODEL, &obj)
	}
	if str, err := ret.EnCode(appConfig.GID, appConfig.GID, obj.Seq, err); err == nil {
		GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_MODEL_ACK), str, 0, ToCloud)
	}
}
func HandleTpQLog(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_MutilFileObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		//读文件内容
		ReadMutilFileContent(protocol.TP_GW_LOG, obj.Data.Id, &ret)
		if str, err := ret.EnCode(appConfig.GID, obj.Seq); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_LOG_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpRLog(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	var ret protocol.Pack_Ack
	var err error
	if err = obj.DeCode(m.PayloadString()); err == nil {
		rd, _ := os.ReadDir(util.GetPath(3))
		for _, fi := range rd {
			if ok := strings.HasSuffix(fi.Name(), ".log"); ok {
				err = os.Remove(util.GetPath(3) + fi.Name())
			}
		}
		if str, err := ret.EnCode(appConfig.GID, appConfig.GID, obj.Seq, err); err == nil {
			GetMQTTMgr().Publish(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_REMOVE_LOG_ACK), str, 0, ToCloud)
		}
	}
}
func HandleTpQSys(m mqtt.Message) {
	var obj protocol.Pack_IDObject
	if err := obj.DeCode(m.PayloadString()); err == nil {
		go SysInfoStat(obj.Seq)
	}
}

type MqttOnline struct {
}

func (o *MqttOnline) GetOnlineMsg() (string, string) {
	//发布上线消息
	var obj protocol.Pack_IDObject
	str, err := obj.EnCode(appConfig.GID, GetNextUint64(), 0)
	if err != nil {
		return "", ""
	}
	return GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_ONLINE), str
}

func (o *MqttOnline) GetWillMsg() (string, string) {
	payload, _ := (&protocol.Pack_IDObject{}).EnCode(appConfig.GID, GetNextUint64(), 0) //遗嘱消息
	return GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_WILL), payload
}

// InitCloudMqttSubscribeTopics 初始化网关级别的主题订阅及路由
func InitCloudMqttSubscribeTopics() {
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_APP), mqtt.AtMostOnce, HandleTpQApp, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_APP), mqtt.AtMostOnce, HandleTpWApp, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SERIAL), mqtt.AtMostOnce, HandleTpQSerial, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_SERIAL), mqtt.AtMostOnce, HandleTpWSerial, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_RTU), mqtt.AtMostOnce, HandleTpQRtu, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_RTU), mqtt.AtMostOnce, HandleTpWRtu, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_MODEL), mqtt.AtMostOnce, HandleTpQModel, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SET_MODEL), mqtt.AtMostOnce, HandleTpWModel, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_LOG), mqtt.AtMostOnce, HandleTpQLog, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_REMOVE_LOG), mqtt.AtMostOnce, HandleTpRLog, ToCloud)
	GetMQTTMgr().Subscribe(GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_SYS), mqtt.AtMostOnce, HandleTpQSys, ToCloud)
}