From 53d80c839ad2aad0a0482ca43fca1fc3a95b9d1c Mon Sep 17 00:00:00 2001
From: zhangqian <zhangqian@123.com>
Date: 星期五, 03 十一月 2023 17:30:54 +0800
Subject: [PATCH] 更改路由

---
 crontask/cron_task.go |   95 ++++++++++++++++++++++++++++++++++-------------
 1 files changed, 69 insertions(+), 26 deletions(-)

diff --git a/crontask/cron_task.go b/crontask/cron_task.go
index 2bdb2c2..de593c3 100644
--- a/crontask/cron_task.go
+++ b/crontask/cron_task.go
@@ -3,11 +3,14 @@
 import (
 	"apsClient/conf"
 	"apsClient/constvar"
+	"apsClient/model"
 	"apsClient/model/common"
 	"apsClient/nsq"
 	"apsClient/pkg/ecode"
 	"apsClient/pkg/logx"
+	"apsClient/serf"
 	"apsClient/service"
+	"apsClient/service/problem"
 	"fmt"
 	"github.com/go-co-op/gocron"
 	"github.com/spf13/cast"
@@ -16,7 +19,7 @@
 
 var s *gocron.Scheduler
 
-func StartTask() error {
+func StartTask(isMaster bool) error {
 	finishNumberTimeInterval := conf.Conf.PLC.FinishNumberTimeInterval
 	totalNumberTimeInterval := conf.Conf.PLC.TotalNumberTimeInterval
 	if finishNumberTimeInterval == 0 {
@@ -75,32 +78,27 @@
 
 	})
 
-	s.Every(60).Seconds().Do(SyncProductionProgress) //鍚屾鐢熶骇鏁版嵁
-	s.Every(30).Seconds().Do(SyncTaskStatus)         //鍚屾浠诲姟鐘舵��
+	if isMaster {
+		s.Every(60).Seconds().Do(SyncProductionProgress) //鍚屾鐢熶骇鏁版嵁
+		s.Every(30).Seconds().Do(SyncTaskStatus)         //鍚屾浠诲姟鐘舵��
+		s.Every(10).Seconds().Do(CheckNsqConn)           //鏌ヨnsq杩炴帴
+
+	}
+
+	s.Every(20).Seconds().Do(QueryClusterStatus) //鏌ヨ闆嗙兢鑺傜偣鏁伴噺
+	s.Every(30).Seconds().Do(ProblemCheck)       //闂璇婃柇
+
 	s.StartAsync()
 	return nil
 }
 
 func SyncProductionProgress() {
-	plcConfig, code := service.NewDevicePlcService().GetDevicePlc()
-	if code != ecode.OK {
+	progressList, err := service.NewProgressService().GetProgressList()
+	if err != nil {
+		logx.Errorf("SyncProductionProgress get records err:%v", err)
 		return
 	}
-	var channels []int32
-	for _, item := range plcConfig.Details {
-		if item.FieldName == constvar.PlcStartAddressTypeFinishNumber {
-			channels = append(channels, item.Channel)
-		}
-	}
-	for _, channel := range channels {
-		progress, err := service.NewProgressService().GetCurrentProgress(channel)
-		if err != nil {
-			return
-		}
-		if progress == nil {
-			return
-		}
-
+	for _, progress := range progressList {
 		caller := nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicSyncTaskProgress, conf.Conf.NsqConf.NodeId), "")
 		err = caller.Send(progress)
 		if err != nil {
@@ -114,6 +112,7 @@
 	if err != nil {
 		logx.Errorf("SyncTaskStatus get records err:%v", err)
 	}
+	syncOkIds := make([]uint, 0, len(records))
 	for _, record := range records {
 		msg := &common.MsgTaskStatusUpdate{
 			WorkOrderId:  record.WorkOrderId,
@@ -127,26 +126,70 @@
 		if err != nil {
 			logx.Errorf("sync task status send msg error:%v", err.Error())
 		}
+		syncOkIds = append(syncOkIds, record.ID)
+	}
+	if len(syncOkIds) > 0 {
+		err = service.NewTaskService().RemoveTaskStatusSync(syncOkIds)
+		if err != nil {
+			logx.Errorf("sync task status delete sync ok records error:%v", err)
+		}
 	}
 }
 
-func StopTask() {
+func RestartTask(isMaster bool) error {
 	if s != nil {
 		s.Stop()
 	}
+	return StartTask(isMaster)
 }
 
 // Once 涓�娆℃�т换鍔�
-func Once() {
+func Once(isMaster bool) {
+	if !isMaster {
+		return
+	}
 	msg := &common.MsgPullDataRequest{DataType: common.PullDataTypeProcessModel}
 	caller := nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicPullDataRequest, conf.Conf.NsqConf.NodeId), constvar.NsqTopicPullDataResponse)
 	err := caller.Send(msg)
 	if err != nil {
-		logx.Errorf("send pull data msg error:%v", err.Error())
+		logx.Errorf("send pull data msg error:%v, msg:%+v", err.Error(), msg)
 	}
-	caller = nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicGetPlcAddress, conf.Conf.NsqConf.NodeId), "")
-	err = caller.Send(common.RequestPlcAddress{DeviceId: conf.Conf.System.DeviceId})
+	msg = &common.MsgPullDataRequest{DataType: common.PullDataTypeDevice}
+	caller = nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicPullDataRequest, conf.Conf.NsqConf.NodeId), constvar.NsqTopicPullDataResponse)
+	err = caller.Send(msg)
 	if err != nil {
-		logx.Infof("get plc address err: %v", err.Error())
+		logx.Errorf("send pull data msg error:%v, msg:%+v", err.Error(), msg)
+	}
+}
+
+func QueryClusterStatus() {
+	clusterStatus, nodeQuantity := serf.QueryClusterStatusAndNodeQuantity()
+	conf.Conf.SerfClusterStatus = clusterStatus
+	conf.Conf.ClusterNodeQuantity = nodeQuantity
+}
+
+func ProblemCheck() {
+	problem.Check()
+}
+
+func CheckNsqConn() {
+	var err error
+	var status constvar.SystemStatusValue
+	if nsq.Ping() {
+		status = constvar.SystemStatusValueNormal
+	} else {
+		status = constvar.SystemStatusValueUnNormal
+	}
+	old, err := model.NewSystemStatusSearch().SetKey(constvar.SystemStatusKeyNsq).First()
+	if err != nil {
+		logx.Errorf("get nsq status err:%v", err)
+		return
+	}
+	if old.Value == status {
+		return
+	}
+	err = model.NewSystemStatusSearch().SetKey(constvar.SystemStatusKeyNsq).Updates(map[string]interface{}{"value": status})
+	if err != nil {
+		logx.Errorf("update nsq status err:%v", err)
 	}
 }

--
Gitblit v1.8.0