喷雾设备离线检查

equipment/spray_equipment.go

@@ -1,25 +1,41 @@
 package equipment
 
+import (
+	"sync"
+	"time"
+)
+
+const (
+	ConnStatusUnknown uint8 = iota
+	ConnStatusOnline
+	ConnStatusOffline
+)
+
 func NewSprayEquipment(id string) *SprayEquipment {
 	return &SprayEquipment{
-		id:        id,
-		speed1:    0,
-		speed2:    0,
-		yv1:       0,
-		yv2:       0,
-		isRunning: false,
-		status:    0,
+		id:           id,
+		speed1:       0,
+		speed2:       0,
+		yv1:          0,
+		yv2:          0,
+		isRunning:    false,
+		errCode:      0,
+		connStatus:   ConnStatusUnknown,
+		connLostTime: 0,
 	}
 }
 
 type SprayEquipment struct {
-	id        string // 设备ID
-	speed1    uint8  // 电磁阀1对应的喷雾量(百分比): 0-100
-	speed2    uint8  // 电磁阀2对应的喷雾量(百分比): 0-100
-	yv1       uint8  // 电磁阀1的状态: 0 停止, 1 启动
-	yv2       uint8  // 电磁阀2的状态: 0 停止, 1 启动
-	isRunning bool   // 是否运行中
-	status    uint8  // 连接状态: 0 未知, 1 在线, 2 离线
+	id             string        // 设备ID
+	speed1         uint8         // 电磁阀1对应的喷雾量(百分比): 0-100
+	speed2         uint8         // 电磁阀2对应的喷雾量(百分比): 0-100
+	yv1            uint8         // 电磁阀1的状态: 0 停止, 1 启动
+	yv2            uint8         // 电磁阀2的状态: 0 停止, 1 启动
+	isRunning      bool          // 是否已启动
+	errCode        uint8         // 错误代码: 0 无, 1 急停, 2 低水位报警
+	connStatus     uint8         // 连接状态: 0 未知, 1 在线, 2 离线
+	connLostTime   time.Duration // 掉线时间
+	connLostTimeMu sync.RWMutex
 }
 
 func (s *SprayEquipment) GetId() string {
@@ -66,6 +82,41 @@ func (s *SprayEquipment) GetIsRunning() bool {
 	return s.isRunning
 }
 
-func (s *SprayEquipment) GetStatus() uint8 {
-	return s.status
+func (s *SprayEquipment) SetErrCode(errCode uint8) {
+	s.errCode = errCode
+}
+
+func (s *SprayEquipment) GetErrCode() uint8 {
+	return s.errCode
+}
+
+func (s *SprayEquipment) SetConnStatus(status uint8) {
+	s.connStatus = status
+}
+
+func (s *SprayEquipment) GetConnStatus() uint8 {
+	return s.connStatus
+}
+
+func (s *SprayEquipment) GetConnLostTime() time.Duration {
+	s.connLostTimeMu.RLock()
+	defer s.connLostTimeMu.RUnlock()
+
+	return s.connLostTime
+}
+
+func (s *SprayEquipment) IncreaseConnLostTime(d time.Duration) time.Duration {
+	s.connLostTimeMu.Lock()
+	defer s.connLostTimeMu.Unlock()
+
+	s.connLostTime = s.connLostTime + d
+	return s.connLostTime
+}
+
+func (s *SprayEquipment) ResetConnLostTime() {
+	s.connLostTimeMu.Lock()
+	defer s.connLostTimeMu.Unlock()
+
+	s.connStatus = ConnStatusOnline
+	s.connLostTime = 0
 }

main.go

@@ -76,10 +76,35 @@ func main() {
 
 	RetrySubscribe:
 		subTopic := mqttx.SubscribeTopic{
-			Name:           "testtopic/PLC325454756",
-			Qos:            0,
-			MessageHandler: func(c mqtt.Client, m mqtt.Message) {},
-			WaitTimeout:    5 * time.Second,
+			Name: "testtopic/PLC325454756",
+			Qos:  0,
+			MessageHandler: func(c mqtt.Client, msg mqtt.Message) {
+				var payload struct {
+					State  uint8 `json:"state"`
+					Speed1 uint8 `json:"Speed1"`
+					Speed2 uint8 `json:"Speed2"`
+					Yv1    uint8 `json:"YV1"`
+					Yv2    uint8 `json:"YV2"`
+					Err    uint8 `json:"ERROR"`
+				}
+
+				if err := json.Unmarshal(msg.Payload(), &payload); err != nil {
+					log.WithFields(log.Fields{
+						"payload": string(msg.Payload()),
+						"error":   err,
+					}).Error("mqtt: 订阅内容解析失败")
+					return
+				}
+
+				sprayEquipment.ResetConnLostTime()
+				sprayEquipment.SetSpeed1(payload.Speed1)
+				sprayEquipment.SetSpeed2(payload.Speed2)
+				sprayEquipment.SetYv1(payload.Yv1)
+				sprayEquipment.SetYv2(payload.Yv2)
+				sprayEquipment.SetIsRunning(payload.State == 1)
+				sprayEquipment.SetErrCode(payload.Err)
+			},
+			WaitTimeout: 5 * time.Second,
 		}
 		if err := c.Subscribe(subTopic); err != nil {
 			log.WithFields(log.Fields{
@@ -90,6 +115,18 @@ func main() {
 			goto RetrySubscribe
 		}
 
+		go func() {
+			d := 5 * time.Second
+
+			for {
+				time.Sleep(d)
+				if sprayEquipment.IncreaseConnLostTime(d) >= 60*time.Second && sprayEquipment.GetConnStatus() != equipment.ConnStatusOffline {
+					sprayEquipment.SetConnStatus(equipment.ConnStatusOffline)
+					log.Warning("喷雾设备已离线")
+				}
+			}
+		}()
+
 		pubTopic := mqttx.PublishTopic{
 			Name:        "testtopic/325454756",
 			Qos:         0,
@@ -104,7 +141,7 @@ func main() {
 					"error": err,
 				}).Error("mqtt: 话题发送失败")
 			}
-			<-time.After(5 * time.Second)
+			time.Sleep(5 * time.Second)
 		}
 	}()
 
@@ -122,7 +159,8 @@ func StatusHandler(c *gin.Context) {
 		"yv1":        sprayEquipment.GetYv1(),
 		"yv2":        sprayEquipment.GetYv2(),
 		"is_running": sprayEquipment.GetIsRunning(),
-		"status":     sprayEquipment.GetStatus(),
+		"error":      sprayEquipment.GetErrCode(),
+		"status":     sprayEquipment.GetConnStatus(),
 	})
 }