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]收到新消息[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 {
		dtuClient, ok := iotMap.MapDtuClient.Get(dtuMsg.SN)
		if !ok {
			dtu, ok2 := iotMap.MapDtuDevice.Get(dtuMsg.SN)
			if !ok2 {
				logger.Errorf("设备未注册: %s", dtu.SN)
			}
			dtuClient = &iotStruct.DtuClient{
				Logger: dtu.Logger,
				Config: dtu.Config,
			}
		}
		conn.SetClientRegisterID(dtuMsg.SN)
		dtuClient.FD = conn
		dtuClient.Online = true
		dtuClient.UpdateTime()
		if !ok || dtuClient.FD.GetClientHost() != dtuClient.FD.GetClientHost() {
			msg := &iotStruct.DtuRegisterChanMsg{
				SN:    dtuMsg.SN,
				Value: dtuClient,
			}
			DtuRegisterChan <- msg
		} else {
			dtuClient.Logger.Infof("== HeartBeat == %s", conn.GetClientHost())
			iotMap.MapDtuClient.Add(dtuMsg.SN, dtuClient)
		}
		if dtuMsg.Cmd == iotStruct.CmdRegister {
			logger.Debugf("== Register ==  %s %s", dtuMsg.SN, conn.GetClientHost())
		} else if dtuMsg.Cmd == iotStruct.CmdHeartBeat {
			logger.Debugf("== HeartBeat == %s %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)
			client.Logger.Debugf("设备离线：%s", addr)
		}
	}
}

func runTran(msg <-chan *iotStruct.DtuRegisterChanMsg) {
	lastMinute := time.Now().Minute() //最后更新时间
	ticker := time.NewTicker(time.Second * 5)
	defer ticker.Stop()
	for {
		select {
		case msg := <-msg:
			dtuClientOnline(msg)
		case nowTime := <-ticker.C:
			lastMinute = collectAndReport(nowTime, lastMinute)
		}
	}
}

// 设备上线更新配置
func dtuClientOnline(msg *iotStruct.DtuRegisterChanMsg) {
	if dtu, ok := iotMap.MapDtuDevice.Get(msg.SN); ok {
		msg.Value.Config = dtu.Config
		msg.Value.Logger = dtu.Logger
		iotMap.MapDtuClient.Add(msg.SN, msg.Value)
		// 设备上线
		UpdateDtuStatus(msg.SN, constant.IotDeviceOnline)
		dtu.Logger.Infof("设备上线：%s ;配置:  %v 从机配置：%v", msg.Value.FD.GetClientHost(), dtu.Config, dtu.Config.SlaveConfig)
	} else {
		logger.Errorf("【设备上线失败】[%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)

		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 {
				dtu.Logger.Errorf("【在线设备更新配置】设备不存在  SN：%s", dtuSn)
				return true
			}
			v.Config = dtu.Config
			v.Logger = dtu.Logger
			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 {
					client.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
	}
}