Device/SERVER/Meter_Service/service/downStreamService/service.go

package downStreamService

import (
	"MeterService/core/config"
	"MeterService/core/logger"
	"MeterService/core/tcpserver"
	"MeterService/data"
	"MeterService/dataStruct"
	"encoding/json"
	"fmt"
	"log"
	"time"
)

func NewDownStreamService() {
	logger.Info("启动DTU服务:%d", config.C.Vber.CmdPort)
	server := tcpserver.New(fmt.Sprintf(":%d", config.C.Vber.CmdPort))
	server.OnNewClient(func(conn *tcpserver.Client) {
		logger.Info("新设备连接：%s", conn.GetClientHost())
	})
	server.OnNewMessage(func(conn *tcpserver.Client, msg []byte) {
		//logger.Debug("[%s]收到新消息：%s", conn.GetClientHost(), string(msg))
		logger.Debug("[%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(data.TranSN)

	server.Listen()
}

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

// 处理DTU注册包及心跳包
func dealDtuRegisterHeartBeat(conn *tcpserver.Client, dtuMsg *dataStruct.DTUManageCmd) {
	if dtuMsg.Cmd == 0 || dtuMsg.Cmd == 1 {
		clientState, _ := data.DtuMapState.Get(dtuMsg.SN)
		conn.SetClientRegisterID(dtuMsg.SN)
		clientState.FD = conn
		clientState.Online = true
		clientState.UpdateTime()
		data.DtuMapState.Add(dtuMsg.SN, clientState)
		data.IpSN.Add(conn.GetClientHost(), dtuMsg.SN)
		onlineSn, ok := data.OnlineSN.Get(dtuMsg.SN)
		if !ok || onlineSn.FD.GetClientHost() != clientState.FD.GetClientHost() {
			msg := &dataStruct.DtuRegisterChanMsg{
				Sn:    dtuMsg.SN,
				Value: clientState,
			}
			data.TranSN <- msg
		}
		if dtuMsg.Cmd == 0 {
			logger.Info(dtuMsg.SN + "[" + clientState.FD.GetClientHost() + "] Register")
		}
		if dtuMsg.Cmd == 1 {
			logger.Info(dtuMsg.SN + "[" + clientState.FD.GetClientHost() + "] HeartBeat")
		}
	} else {
		logger.Error("Register || HeartBeat 失败")
	}
}

// 处理DTU断开连接
func dealDTUDisconnect(conn *tcpserver.Client) {
	sn := conn.GetClientRegisterID()
	addr := conn.GetClientHost()
	logger.Info("[%s]设备断开连接：%s", addr, sn)
	if sn != "" {
		if client, ok := data.DtuMapState.Get(sn); ok && client.FD.GetClientHost() == addr {
			client.Online = false
			data.DtuMapState.Add(sn, client)
			logger.Debug("[%s]设备离线：%s", addr, sn)
		}
		if client, ok := data.OnlineSN.Get(sn); ok && client.FD.GetClientHost() == addr {
			data.OnlineSN.Remove(sn)
		}
	}
}

func runTran(msg <-chan *dataStruct.DtuRegisterChanMsg) {
	lastMinute := time.Now().Minute() //最后更新时间
	ticker := time.NewTicker(time.Second * 5)
	defer ticker.Stop()
	for {
		select {
		case msg := <-msg:
			logger.Info("【更新设备配置】 SN：%s", msg.Sn)
			updateOnlineConf(msg)
		case nowTime := <-ticker.C:
			minute := nowTime.Minute() //当前分钟
			if minute != lastMinute {
				count := 0
				onlineDtu := make(map[string]*dataStruct.ClientState)

				for _, v := range data.DtuMap.Map.Map {
					vv := v.(*dataStruct.DTUDeviceState)
					log.Println("===============", vv.Info.Name)
				}
				data.OnlineSN.Map.Range(func(key, value interface{}) bool {
					v := value.(*dataStruct.ClientState)
					dtuSn := key.(string)
					logger.Debug("【检查在线设备配置】 SN：%s", dtuSn)
					if v.Config != nil {
						onlineDtu[dtuSn] = v
					}
					count++
					return true
				})
				if count == 0 {
					logger.Info("没有在线设备。")
				}
				for dtuSn, client := range onlineDtu {
					if minute%client.Config.Secs == 0 {
						// 采集上报数据
						if dataArray := collectData(dtuSn, client.Config); len(dataArray) > 0 {
							logger.Debug("开始上报平台：%s", dtuSn)
							reportData(dataArray, client.Config)
						}
					}
				}

				lastMinute = minute
			}
		}
	}
}

// 更新设备配置
func updateOnlineConf(msg *dataStruct.DtuRegisterChanMsg) {
	dtuConfig := &dataStruct.DtuConfig{}
	if dtu, ok := data.DtuMap.Get(msg.Sn); ok {
		if err := json.Unmarshal([]byte(dtu.Info.Configs), dtuConfig); err != nil {
			logger.Error("[%s]设备配置解析失败。", msg.Sn)
			return
		}
		msg.Value.Config = dtuConfig
		data.OnlineSN.Add(msg.Sn, msg.Value)
		logger.Debug("[%s]设备配置更新成功。 %v ", msg.Sn, *dtuConfig)
	} else {
		logger.Debug("【更新设备配置失败】[%s]设备不存在。", msg.Sn)
	}
}