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"
	"time"
)

var (
	errCount = 0
)

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 == dataStruct.CmdRegister || dtuMsg.Cmd == dataStruct.CmdHeartBeat {
		//clientState, _ := data.DtuMapState.Get(dtuMsg.SN)
		clientState, ok := data.OnlineSN.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 := &dataStruct.DtuRegisterChanMsg{
				Sn:    dtuMsg.SN,
				Value: clientState,
			}
			data.TranSN <- msg
		} else {
			logger.Debug("[%s]设备上线：%s", clientState.FD.GetClientHost(), dtuMsg.SN)
			data.OnlineSN.Add(dtuMsg.SN, clientState)
		}
		if dtuMsg.Cmd == dataStruct.CmdRegister {
			logger.Info("======【%s】====== Register  %s", dtuMsg.SN, conn.GetClientHost())
		}
		if dtuMsg.Cmd == dataStruct.CmdHeartBeat {
			logger.Info("======【%s】====== HeartBeat %s", dtuMsg.SN, conn.GetClientHost())
		}
	} 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)
			//client.Online = false
			//data.OnlineSN.Add(sn, client)
			logger.Debug("[%s]设备离线：%s", addr, 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:
			lastMinute = collectAndReport(nowTime, lastMinute)
		}
	}
}

// 设备上线更新配置
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]设备上线：%s ;配置:  %v", msg.Value.FD.GetClientHost(), msg.Sn, *dtuConfig)
	} else {
		logger.Debug("【设备上线失败】[%s]设备不存在。", msg.Sn)
	}
}

// 采集并上报数据
func collectAndReport(nowTime time.Time, lastMinute int) int {
	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)
			dtu, ok := data.DtuMap.Get(dtuSn)
			if !ok || dtu.Info.Configs == "" {
				logger.Error("【在线设备更新配置】设备不存在  SN：%s", dtuSn)
				return true
			}
			dtuConfig := &dataStruct.DtuConfig{}
			if err := json.Unmarshal([]byte(dtu.Info.Configs), dtuConfig); err != nil {
				logger.Error("【在线设备更新配置】解析失败  SN：%s", dtuSn)
				return true
			}
			v.Config = dtuConfig
			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.Secs == 0 {
				// 采集上报数据
				if dataArray := collectData(dtuSn, client.Config); len(dataArray) > 0 {
					logger.Debug("开始上报平台：%s", dtuSn)
					reportData(dataArray, client.Config)
				}
			}
		}
		errCount = 0
	}
	return minute

}