物联网软件

iot/EdgeAgent/main/mqtt_service.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"crypto/tls"
+	"sentinel/pkg/config"
 	"sentinel/pkg/log"
 	"time"
 
@@ -9,8 +10,9 @@ import (
 )
 
 type MQTTService struct {
-	client  mqtt.Client
-	handler func(topic string, payload []byte)
+	client        mqtt.Client
+	handler       func(topic string, payload []byte)
+	subscriptions map[string]struct{} // 记录已订阅 topic
 }
 
 func NewMQTTService(broker, clientID, username, password string, keepalive int) *MQTTService {
@@ -19,20 +21,45 @@ func NewMQTTService(broker, clientID, username, password string, keepalive int)
 	opts.SetClientID(clientID)
 	opts.SetUsername(username)
 	opts.SetPassword(password)
+	opts.AutoReconnect = true
+	opts.OnReconnecting = func(c mqtt.Client, opts *mqtt.ClientOptions) {
+		log.Println("正在重连到物联网服务器...")
+	}
 	opts.SetKeepAlive(time.Duration(keepalive) * time.Second)
 	opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
-
-	ms := &MQTTService{}
+	// 设置遗嘱消息
+	lwtTopic, lwtPayload := getStatusInfoTopicPayload(false)
+	opts.SetWill(
+		lwtTopic,   // topic
+		lwtPayload, // payload
+		1,          // qos
+		false,      // retained
+	)
+	ms := &MQTTService{subscriptions: make(map[string]struct{})}
 	opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) {
 		if ms.handler != nil {
 			ms.handler(m.Topic(), m.Payload())
 		}
 	})
 	opts.OnConnect = func(c mqtt.Client) {
-		log.Println("物联网服务已连接")
+		log.Println("物联网服务器已连接")
+		cfg := config.LoadConfig()
+		cfg.ControlState = "CONNECTED"
+		cfg.LastAliveAt = time.Now().Unix()
+		config.WriteLocalConfig(cfg)
+		// 连接成功 发送状态信息
+		topic, payload := getStatusInfoTopicPayload(true)
+		ms.PublicMsg(topic, payload, false)
 	}
 	opts.OnConnectionLost = func(c mqtt.Client, err error) {
-		log.Println("物联网服务已断开：", err)
+		log.Println("物联网服务器已断开：", err)
+		cfg := config.LoadConfig()
+		cfg.ControlState = "DEGRADED"       // 表示短暂断开
+		cfg.LastAliveAt = time.Now().Unix() // 更新这个时间戳，让 daemon 与 主程序看到尚可忍耐
+
+		if err := config.WriteLocalConfig(cfg); err != nil {
+			log.Println("[main] 写状态失败:", err)
+		}
 	}
 
 	ms.client = mqtt.NewClient(opts)
@@ -52,7 +79,7 @@ func (m *MQTTService) Subscribe(topic string, qos byte) error {
 	if token.Wait() && token.Error() != nil {
 		return token.Error()
 	}
-	log.Println("物联网服务消息订阅：", topic)
+	log.Debug("物联网服务消息订阅：", topic)
 	return nil
 }
 
@@ -71,3 +98,48 @@ func (m *MQTTService) Close() {
 		m.client.Disconnect(250)
 	}
 }
+
+func (m *MQTTService) UnsubscribeTopic(topic string) error {
+	token := m.client.Unsubscribe(topic)
+	go func() {
+		if token.Wait() && token.Error() != nil {
+			log.Println("取消订阅失败:", token.Error())
+		} else {
+			delete(m.subscriptions, topic)
+			log.Debug("取消订阅成功:", topic)
+		}
+	}()
+	return nil
+}
+
+// UnsubscribeAll 取消所有已订阅 topic
+func (m *MQTTService) UnsubscribeAll() {
+	for topic := range m.subscriptions {
+		_ = m.client.Unsubscribe(topic)
+		delete(m.subscriptions, topic)
+	}
+}
+
+// SubscribeTopic 订阅指定 topic，并记录可取消
+func (m *MQTTService) SubscribeTopic(topic string, qos byte) error {
+	token := m.client.Subscribe(topic, qos, nil)
+	go func() {
+		if token.Wait() && token.Error() != nil {
+			log.Println("订阅失败:", token.Error())
+		} else {
+			m.subscriptions[topic] = struct{}{}
+			log.Debug("订阅成功:", topic)
+		}
+	}()
+	return nil
+}
+
+func (m *MQTTService) PublicMsg(topic string, payload string, retained bool) {
+	qos := byte(1)
+	log.Debug("发送消息：", topic)
+	if err := m.Publish(topic, qos, retained, payload); err != nil {
+		log.Println("发送信息出错：", err)
+	} else {
+		log.Debug("发送信息：", string(payload))
+	}
+}