IOT/SERVER/IotAdmin/iot/service/downService/service.go

package iotDownService

import (
	"IotAdmin/common/global"
	"IotAdmin/core/logger"
	"IotAdmin/core/sdk"
	"IotAdmin/core/sdk/config"
	"IotAdmin/core/tcpserver"
	"IotAdmin/iot/constant"
	iotMap "IotAdmin/iot/map"
	iotStruct "IotAdmin/iot/struct"
	"fmt"
	"time"
)

var (
	errCount        = 0
	DtuRegisterChan chan *iotStruct.DtuRegisterChanMsg
)

func NewDownStreamService() {
	logger.Infof("启动DTU服务:%d", config.ApplicationConfig.TcpPort)
	server := tcpserver.New(fmt.Sprintf(":%d", config.ApplicationConfig.TcpPort))
	server.OnNewClient(func(conn *tcpserver.Client) {
		logger.Infof("新设备连接：%s", conn.GetClientHost())
	})
	server.OnNewMessage(func(conn *tcpserver.Client, msg []byte) {
		//logger.Debugf("[%s]收到新消息：%s", conn.GetClientHost(), string(msg))
		logger.Debugf("[%s]收到新消息[byte]：%v", conn.GetClientHost(), msg)
		if dealDtuUpMsg(conn, msg) {
			conn.PutRawRes(msg)
		}
	})
	server.OnClientConnectionClosed(func(conn *tcpserver.Client, err error) {
		dealDTUDisconnect(conn)
	})

	go runTran(DtuRegisterChan)

	server.Listen()
}

// 处理DTU上报数据
func dealDtuUpMsg(conn *tcpserver.Client, msg []byte) bool {
	var (
		dtuMsg = &iotStruct.DTUManageCmd{}
	)
	dtuMsg.DmcUnmarshal(msg)
	if dtuMsg.Cmd == iotStruct.CmdCollect {
		return true
	} else {
		dealDtuRegisterHeartBeat(conn, dtuMsg)
	}
	return false
}

// 处理DTU注册包及心跳包
func dealDtuRegisterHeartBeat(conn *tcpserver.Client, dtuMsg *iotStruct.DTUManageCmd) {
	if dtuMsg.Cmd == iotStruct.CmdRegister || dtuMsg.Cmd == iotStruct.CmdHeartBeat {
		//clientState, _ := data.DtuMapState.Get(dtuMsg.SN)
		clientState, ok := iotMap.MapDtuClient.Get(dtuMsg.SN)
		conn.SetClientRegisterID(dtuMsg.SN)
		clientState.FD = conn
		clientState.Online = true
		clientState.UpdateTime()

		//data.IpSN.Add(conn.GetClientHost(), dtuMsg.SN)
		//onlineSn, ok := data.OnlineSN.Get(dtuMsg.SN)
		if !ok || clientState.FD.GetClientHost() != clientState.FD.GetClientHost() {
			msg := &iotStruct.DtuRegisterChanMsg{
				Sn:    dtuMsg.SN,
				Value: clientState,
			}
			DtuRegisterChan <- msg
		} else {
			logger.Debugf("[%s]设备上线：%s", clientState.FD.GetClientHost(), dtuMsg.SN)
			iotMap.MapDtuClient.Add(dtuMsg.SN, clientState)
		}
		if dtuMsg.Cmd == iotStruct.CmdRegister {
			// 设备上线
			UpdateDtuStatus(dtuMsg.SN, constant.IotDeviceOnline)
			logger.Infof("======【%s】====== Register  %s", dtuMsg.SN, conn.GetClientHost())
		}
		if dtuMsg.Cmd == iotStruct.CmdHeartBeat {
			logger.Infof("======【%s】====== HeartBeat %s", dtuMsg.SN, conn.GetClientHost())
		}
	} else {
		logger.Errorf("Register || HeartBeat 失败")
	}
}

// 处理DTU断开连接
func dealDTUDisconnect(conn *tcpserver.Client) {
	sn := conn.GetClientRegisterID()
	addr := conn.GetClientHost()
	logger.Infof("[%s]设备断开连接：%s", addr, sn)
	if sn != "" {
		UpdateDtuStatus(sn, constant.IotDeviceOffline)
		if client, ok := iotMap.MapDtuClient.Get(sn); ok && client.FD.GetClientHost() == addr {
			iotMap.MapDtuClient.Remove(sn)
			logger.Debugf("[%s]设备离线：%s", addr, sn)
		}
	}
}

func runTran(msg <-chan *iotStruct.DtuRegisterChanMsg) {
	lastMinute := time.Now().Minute() //最后更新时间
	ticker := time.NewTicker(time.Second * 5)
	defer ticker.Stop()
	for {
		select {
		case msg := <-msg:
			logger.Infof("【更新设备配置】 SN：%s", msg.Sn)
			updateOnlineConf(msg)
		case nowTime := <-ticker.C:
			lastMinute = collectAndReport(nowTime, lastMinute)
		}
	}
}

// 设备上线更新配置
func updateOnlineConf(msg *iotStruct.DtuRegisterChanMsg) {
	if dtu, ok := iotMap.MapDtuDevice.Get(msg.Sn); ok {
		msg.Value.Config = dtu.Config
		iotMap.MapDtuClient.Add(msg.Sn, msg.Value)
		logger.Debugf("[%s]设备上线：%s ;配置:  %v", msg.Value.FD.GetClientHost(), msg.Sn, dtu.Config)
	} else {
		logger.Debugf("【设备上线失败】[%s]设备不存在。", msg.Sn)
	}
}

// 采集并上报数据
func collectAndReport(nowTime time.Time, lastMinute int) int {
	minute := nowTime.Minute() //当前分钟
	if minute != lastMinute {
		count := 0
		onlineDtu := make(map[string]*iotStruct.DtuClient)

		//for _, v := range data.DtuMap.Map.Map {
		//	vv := v.(*iotStruct.DTUDeviceState)
		//	log.Println("===============", vv.Info.Name)
		//}
		iotMap.MapDtuClient.Map.Range(func(key, value interface{}) bool {
			v := value.(*iotStruct.DtuClient)
			dtuSn := key.(string)
			dtu, ok := iotMap.MapDtuDevice.Get(dtuSn)
			if !ok || dtu.Config == nil {
				logger.Errorf("【在线设备更新配置】设备不存在  SN：%s", dtuSn)
				return true
			}

			v.Config = dtu.Config
			onlineDtu[dtuSn] = v
			count++
			return true
		})
		if count == 0 && errCount < 10 {
			errCount++
			logger.Info("没有在线设备。")
			return minute
		}
		for dtuSn, client := range onlineDtu {
			if minute%client.Config.Cycle == 0 {
				// 采集上报数据
				if dataArray := collectData(dtuSn, client.Config); len(*dataArray) > 0 {
					logger.Debugf("开始上报平台：%s", dtuSn)
					reportData(dataArray)
				}
			}
		}
		errCount = 0
	}
	return minute

}

func UpdateDtuStatus(sn string, status int) {
	q := sdk.Runtime.GetMemoryQueue("")
	mp := make(map[string]interface{})
	mp["sn"] = sn
	mp["status"] = status
	mp["time"] = time.Now()
	message, err := sdk.Runtime.GetStreamMessage("", global.DtuChangeStatus, mp)
	if err != nil {
		logger.Errorf("构建更新设备在线状态 message [%s]失败: %v", global.DtuChangeStatus, err)
		return
	}
	err = q.Append(message)
	if err != nil {
		return
	}
}