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/util"
	"fmt"
	"net/rpc"
	"os"
	"runtime"
	"time"
)

var (
	applicationName = "pji-control"
	localStatus     = "idle"
	cloudStatus     = "NONE"
)

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 main() {
	//go judgeState1()
	//go pkg.JudgeLocal()
	//go judgeState3()

	go pkg.GetLocalStatus(&localStatus)

	for {
		time.Sleep(5 * time.Second)
		fmt.Println("localStatus: ", localStatus)
	}
	//select {}
}

func judgeState1() {
	init := true
	turnLength := 60
	lastStatus := "NONE"
	//  轮询任务接口判断是否有更新
	for {
		c_log.GlobalLogger.Errorf("一轮次扫描时间【%v】秒：", turnLength)
		if init {
			time.Sleep(time.Duration(1) * time.Second)
			init = false
		} else {
			time.Sleep(time.Duration(turnLength) * time.Second)
		}
		// 1 获取当前设备的任务的 status
		status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
		if err != nil {
			c_log.GlobalLogger.Error("获取配置status失败：", err)
			continue
		}
		// 2 判断 status
		// UN_CHANGE 没有新的任务，无需更改
		// CHANGE 有新的任务，需要杀死旧的任务并重启
		// NONE 设备没有配置任务，需要杀死旧的任务
		if status == "UN_CHANGE" {
			lastStatus = "UN_CHANGE"
			continue
		} else if status == "CHANGE" || status == "NONE" {
			if lastStatus == "CHANGE" && status == "CHANGE" { // 供更新使用
				commonConfig.InitPlatformConfig()
				continue
			}
			if lastStatus == "NONE" && status == "NONE" {
				continue
			}

			if lastStatus == "NONE" && status == "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)
				lastStatus = status
				c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
				commonConfig.InitPlatformConfig()
				continue
			}

			// 3 发送rpc信号杀死采集程序
			var killArgs commonService.KillSignal
			if lastStatus == "UN_CHANGE" && status == "CHANGE" {
				killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
				c_log.GlobalLogger.Info("更新任务，发送rpc重启信号：", killArgs)
			}
			if lastStatus == "UN_CHANGE" && status == "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 {
				// 此处如果连接失败说明采集程序已经停止了
				lastStatus = "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
			}
			lastStatus = status
			c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
			commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
			if err = KillRpcClient.Close(); err != nil {
				// 不做处理
			}
		} else {
			c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
		}
	}
}
func judgeState3() {}