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 |  127 +++++++++++++++++++++++++++++++++++------
 1 files changed, 107 insertions(+), 20 deletions(-)

diff --git a/crontask/cron_task.go b/crontask/cron_task.go
index 9454dbe..de593c3 100644
--- a/crontask/cron_task.go
+++ b/crontask/cron_task.go
@@ -3,18 +3,23 @@
 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"
 	"time"
 )
 
-func InitTask() error {
+var s *gocron.Scheduler
 
+func StartTask(isMaster bool) error {
 	finishNumberTimeInterval := conf.Conf.PLC.FinishNumberTimeInterval
 	totalNumberTimeInterval := conf.Conf.PLC.TotalNumberTimeInterval
 	if finishNumberTimeInterval == 0 {
@@ -23,7 +28,7 @@
 	if totalNumberTimeInterval == 0 {
 		totalNumberTimeInterval = 60
 	}
-	s := gocron.NewScheduler(time.UTC)
+	s = gocron.NewScheduler(time.UTC)
 	_, err := s.Every(finishNumberTimeInterval).Seconds().Do(func() {
 		plcConfig, code := service.NewDevicePlcService().GetDevicePlc()
 		if code != ecode.OK {
@@ -73,36 +78,118 @@
 
 	})
 
-	s.Every(180).Seconds().Do(SyncProductionProgress) //鍚屾鐢熶骇鏁版嵁
+	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 {
 			logx.Errorf("SyncProductionProgress error:%v", err.Error())
 		}
 	}
+}
 
+func SyncTaskStatus() {
+	records, err := service.NewTaskService().GetTaskStatusSync(100)
+	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,
+			ProcedureID:  record.ProcedureID,
+			DeviceId:     record.DeviceId,
+			IsProcessing: record.IsProcessing,
+			IsFinish:     record.IsFinish,
+		}
+		caller := nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicTaskProcedureStatusUpdate, conf.Conf.NsqConf.NodeId), "")
+		err = caller.Send(msg)
+		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 RestartTask(isMaster bool) error {
+	if s != nil {
+		s.Stop()
+	}
+	return StartTask(isMaster)
+}
+
+// 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, msg:%+v", err.Error(), msg)
+	}
+	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.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