cicv_data_closed_loop/aarch64/pjibot_guide/control/main.go

package main

import (
	commonConfig "cicv-data-closedloop/aarch64/pjibot_guide/common/config"
	commonService "cicv-data-closedloop/aarch64/pjibot_guide/common/service"
	"cicv-data-closedloop/aarch64/pjibot_guide/common/variable"
	"cicv-data-closedloop/aarch64/pjibot_guide/control/pkg"
	"cicv-data-closedloop/common/config/c_log"
	"cicv-data-closedloop/common/entity"
	"cicv-data-closedloop/common/util"
	"fmt"
	"net/rpc"
	"os"
	"runtime"
	"time"
)

var (
	applicationName = "pji-control"
	localStatus     = "idle"
	cloudStatus     = "NONE"
	lastLocalStatus = "idle"
	lastCloudStatus = "NONE"
	launchedFlag    = false
	renewedFlag     = false
	configuredFlag  = false
	renewTimer      *time.Timer // 续约定时器
	RenewDur        = 5         // 续约时间为5分钟
)

func init() {
	runtime.GOMAXPROCS(1)
	// 初始化日志配置
	c_log.InitLog(variable.LogDir, applicationName)
	// 初始化本地配置文件（第1处配置，在本地文件）
	commonConfig.InitLocalConfig(variable.LocalConfigPath)
	// 初始化Oss连接信息
	commonConfig.InitOssConfig()
	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
	commonConfig.InitCloudConfig()
	// 初始化rpc客户端，用于杀死旧的采集程序

	// 初始化websocket配置
	commonConfig.InitWebsocketConfig()
}

func renew() {
	if renewTimer != nil {
		renewTimer.Stop()
	}
	renewedFlag = true
	renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() {
		renewedFlag = false
	})
	c_log.GlobalLogger.Infof("续约时间【%v】分钟 - 成功。", RenewDur)
}

func startMasterNode() {
	c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
	commonConfig.InitPlatformConfig()

	if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
		c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
		os.Exit(-1)
	}
	c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)

	launchedFlag = true
	c_log.GlobalLogger.Info("数采程序启动 - 成功。")
}

func stopMasterNode() {
	// 发送rpc信号杀死采集程序
	var killArgs commonService.KillSignal
	killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
	c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号：", killArgs)
	KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
	if err != nil {
		// 此处如果连接失败说明采集程序已经停止了
		lastCloudStatus = "NONE"
		c_log.GlobalLogger.Error("采集程序已经停止：", err)
		return
	}

	reply := 0
	if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
		c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错：", err)
		// 这里可能会报错unexpected EOF但是不影响，先注释 close 和 continue
		//KillRpcClient.Close()
		//continue
	}

	c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
	commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
	if err = KillRpcClient.Close(); err != nil {
		// 不做处理
	}

	launchedFlag = false
	c_log.GlobalLogger.Info("数采程序关闭 - 成功。")
}

func main() {
	localTurnLength := 2  // s
	cloudTurnLength := 60 // s
	waitStopLength := 1   // min

	// 更新本地任务状态
	go pkg.GetLocalStatus(&localStatus, &lastLocalStatus, localTurnLength)
	// 更新云端任务状态
	go pkg.GetCloudStatus(&cloudStatus, &lastCloudStatus, cloudTurnLength)

	for {
		if launchedFlag { // 当前已启动master节点
			time.Sleep(time.Duration(cloudTurnLength) * time.Second)
		} else {
			time.Sleep(time.Duration(localTurnLength) * time.Second)
		}

		fmt.Println("localStatus: ", localStatus, "lastLocalStatus: ", lastLocalStatus)
		fmt.Println("cloudStatus: ", cloudStatus, "lastCloudStatus: ", lastCloudStatus)

		// 综合判断 cloudStatus 和 localStatus
		// cloudStatus
		// UN_CHANGE 没有新的任务，无需更改
		// CHANGE 有新的任务，需要杀死旧的数采任务并重启
		// NONE 设备没有配置任务，需要杀死旧的数采任务
		// localStatus
		// idle 空闲状态，此状态下不启动数采任务
		// running 繁忙状态，此状态需要启动数采任务
		// error 错误状态，此状态下不启动数采任务

		// 本地任务状态负责启停master
		if localStatus == "running" {
			if !launchedFlag { // 目前未启动数采程序
				c_log.GlobalLogger.Info("数采程序启动 - 开始。")
				startMasterNode()
			} else { // 已启动数采程序
				c_log.GlobalLogger.Info("设备仍处于运行状态，续约 - 开始。")
				// 续约
				renew()
			}
			continue
		} else if localStatus == "idle" {
			if !renewedFlag && launchedFlag && len(entity.TimeWindowConsumerQueue) == 0 {
				c_log.GlobalLogger.Info("设备不在运行状态且没有待处理的数据，数采程序关闭 - 开始。")
				time.Sleep(time.Duration(waitStopLength) * time.Minute)
				stopMasterNode()
			}
			continue
		} else if localStatus == "error" {
			c_log.GlobalLogger.Error("设备运行状态出错，停止数采程序。")
			stopMasterNode()
		} else {
			c_log.GlobalLogger.Error("未知的设备运行状态。【status】=", localStatus)
		}

		// 云端任务状态负责更新配置
		go pkg.GetCloudConfig(cloudStatus, lastCloudStatus, cloudTurnLength)

		//if cloudStatus == "UN_CHANGE" {
		//	continue
		//} else if cloudStatus == "CHANGE" || cloudStatus == "NONE" {
		//	if lastCloudStatus == "CHANGE" && cloudStatus == "CHANGE" { // 供更新使用
		//		commonConfig.InitPlatformConfig()
		//		continue
		//	}
		//	if lastCloudStatus == "NONE" && cloudStatus == "NONE" {
		//		continue
		//	}
		//
		//	if lastCloudStatus == "NONE" && cloudStatus == "CHANGE" {
		//		if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
		//			c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
		//			os.Exit(-1)
		//		}
		//		c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
		//		c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
		//		commonConfig.InitPlatformConfig()
		//		continue
		//	}
		//
		//	// 3 发送rpc信号杀死采集程序
		//	var killArgs commonService.KillSignal
		//	if lastCloudStatus == "UN_CHANGE" && cloudStatus == "CHANGE" {
		//		killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
		//		c_log.GlobalLogger.Info("更新任务，发送rpc重启信号：", killArgs)
		//	}
		//	if lastCloudStatus == "UN_CHANGE" && cloudStatus == "NONE" {
		//		killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
		//		c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号：", killArgs)
		//	}
		//
		//	KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
		//	if err != nil {
		//		// 此处如果连接失败说明采集程序已经停止了
		//		lastCloudStatus = "NONE"
		//		c_log.GlobalLogger.Error("采集程序已经停止：", err)
		//		continue
		//	}
		//
		//	reply := 0
		//	if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
		//		c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错：", err)
		//		// 这里可能会报错unexpected EOF但是不影响，先注释 close 和 continue
		//		//KillRpcClient.Close()
		//		//continue
		//	}
		//
		//	c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
		//	commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
		//	if err = KillRpcClient.Close(); err != nil {
		//		// 不做处理
		//	}
		//} else {
		//	c_log.GlobalLogger.Error("未知的采集任务状态。【cloudStatus】=", cloudStatus)
		//}
	}
	//select {}
}