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) } }