完善自我更新逻辑

2025-12-30 17:54:33 +08:00
parent ec29d883bd
commit 62e8ecb7d6
12 changed files with 248 additions and 112 deletions
@@ -4,50 +4,85 @@ import (
 	"encoding/json"
 	"os"
 	"os/exec"
-	"path/filepath"
+	"runtime"
+	"sentinel/pkg/config"
 	"sentinel/pkg/log"
 	model2 "sentinel/pkg/model"
 	"sentinel/pkg/utils"
+	"strconv"
+	"syscall"
 	"time"
 )

 type BusinessService struct {
-	mqtt       *MQTTService
-	deviceID   string
-	project    string
-	cmdTopic   string
-	deviceType string
+	mqtt          *MQTTService
+	deviceID      string
+	deptId        string
+	cmdTopic      string
+	deviceType    string
+	subscriptions map[string]struct{} // 记录已订阅 topic
 }

-func NewBusinessService(m *MQTTService, project, deviceType, deviceID string) *BusinessService {
-	// 根据统一规则生成 topic
-	cmdTopic := project + "/cmd/" + deviceType + "/" + deviceID + "/#"
+func NewBusinessService(m *MQTTService, deviceID string) *BusinessService {
 	return &BusinessService{
-		mqtt:       m,
-		project:    project,
-		deviceID:   deviceID,
-		cmdTopic:   cmdTopic,
-		deviceType: deviceType,
+		mqtt:          m,
+		deviceID:      deviceID,
+		subscriptions: make(map[string]struct{}),
 	}
 }

+// SubscribeTopic 订阅指定 topic，并记录可取消
+func (b *BusinessService) SubscribeTopic(topic string, qos byte) error {
+	if err := b.mqtt.Subscribe(topic, qos); err != nil {
+		return err
+	}
+	b.subscriptions[topic] = struct{}{}
+	return nil
+}
+
+func getInitTopic(deviceID string) string {
+	return "+/+/+/" + deviceID + "/#"
+}
+func (b *BusinessService) getOwnTopic(deviceID string) string {
+	return b.deptId + "/cmd/" + b.deviceType + "/" + deviceID + "/#"
+}
+
 func (b *BusinessService) Start() error {
-	// 订阅 cmd topic
-	if err := b.mqtt.Subscribe(b.cmdTopic, 1); err != nil {
+	if err := b.SubscribeTopic(getInitTopic(b.deviceID), 1); err != nil {
 		return err
 	}
+
 	b.mqtt.SetMessageHandler(b.onMQTTMessage)
+
 	// 第一次连接就发送状态信息
 	b.SendStatusInfo()

 	return nil
 }

+// UnsubscribeTopic 取消订阅指定 topic
+func (b *BusinessService) UnsubscribeTopic(topic string) error {
+	token := b.mqtt.client.Unsubscribe(topic)
+	if token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	delete(b.subscriptions, topic)
+	return nil
+}
+
+// UnsubscribeAll 取消所有已订阅 topic
+func (b *BusinessService) UnsubscribeAll() {
+	for topic := range b.subscriptions {
+		_ = b.mqtt.client.Unsubscribe(topic)
+		delete(b.subscriptions, topic)
+	}
+}
+
 // 消息处理
 func (b *BusinessService) onMQTTMessage(topic string, payload []byte) {
 	model := model2.FromStringToMqttTopic(topic)
-	// 是指令
-	if model.Domain == "cmd" {
+	// 指令
+	if model.Domain == "cmd" && model.DeviceType == b.deviceType {
 		log.Println("收到指令：", model.Resource)
 		switch model.Resource {
 		case "ping":
@@ -61,14 +96,26 @@ func (b *BusinessService) onMQTTMessage(topic string, payload []byte) {
 		default:
 			log.Println("未知的命令：", model.Resource)
 		}
+	} else if model.Domain == "status" && model.Resource == "receipt" {
+		b.deviceType = model.DeviceType
+		b.deptId = model.DeptId
+		// 取消订阅之前的初始化主题
+		if b.UnsubscribeTopic(getInitTopic(b.deviceID)) != nil {
+			log.Error("无法取消初始化主题")
+			return
+		}
+		// 新订阅属于自己的主题
+		if b.SubscribeTopic(b.getOwnTopic(b.deviceID), 1) != nil {
+			log.Error("无法定于属于自己的主题")
+			return
+		}
+		log.Println("设备初始化成功：所属项目：", model.DeptId, "\t设备类型：", model.DeviceType)
 	}
 }

 func (b *BusinessService) SendStatusInfo() {
 	info := map[string]interface{}{
-		"project":      utils.PROJECT,
-		"deviceType":   utils.DEVICE_TPYE,
-		"version":      utils.APP_VERSION,
+		"version":      config.APP_VERSION,
 		"online":       true,
 		"ip":           utils.GetLocalIP(),
 		"hostname":     utils.GetHostname(),
@@ -81,14 +128,15 @@ func (b *BusinessService) SendStatusInfo() {
 	}

 	payload, _ := json.Marshal(info)
-	topic := b.project + "/status/" + b.deviceType + "/" + b.deviceID + "/info"
+	topic := "x/status/x/" + b.deviceID + "/info"
 	qos := byte(1)
 	retained := true
+	log.Println("发送消息：", topic)

 	if err := b.mqtt.Publish(topic, qos, retained, payload); err != nil {
-		log.Println("[BUS] failed to send status info:", err)
+		log.Println("发送状态信息出错：", err)
 	} else {
-		log.Println("[BUS] status info sent:", string(payload))
+		log.Println("发送状态信息：", string(payload))
 	}
 }

@@ -107,25 +155,37 @@ func (b *BusinessService) handleRestart() {
 	os.Exit(0)
 }

-// 更新程序
+// handleCheckUpdate 触发更新流程（主程序侧）
 func (b *BusinessService) handleCheckUpdate() {
-	exe, _ := os.Executable()
-	updaterPath := filepath.Join(filepath.Dir(exe), "updater")
-	if _, err := os.Stat(updaterPath); os.IsNotExist(err) {
-		if _, err2 := os.Stat(updaterPath + ".exe"); err2 == nil {
-			updaterPath = updaterPath + ".exe"
-		} else {
-			log.Println("[BUS] updater not found")
-			return
-		}
+
+	args := []string{
+		"--version", strconv.Itoa(config.APP_VERSION),
 	}
-	cmd := exec.Command(updaterPath, "--target", exe)
+
+	cmd := exec.Command("./updater.exe", args...)
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
+
+	// OS 级脱离父进程
+	switch runtime.GOOS {
+	case "windows":
+		cmd.SysProcAttr = &syscall.SysProcAttr{
+			CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP,
+		}
+	}
+
 	if err := cmd.Start(); err != nil {
 		log.Println("[BUS] failed to start updater:", err)
 		return
 	}
-	log.Println("[BUS] exiting main program for update")
+
+	log.Println(
+		"[BUS] updater started (pid=%d), exiting main program\n",
+		cmd.Process.Pid,
+	)
+
+	// 给 updater 留出启动窗口（尤其是 systemd / docker 环境）
+	time.Sleep(500 * time.Millisecond)
+
 	os.Exit(0)
 }
@@ -2,41 +2,49 @@ package main

 import (
 	"fmt"
-	"sentinel/pkg/utils"
 	"time"

+	"sentinel/pkg/config"
 	"sentinel/pkg/device"
 	"sentinel/pkg/log"
 )

 func main() {
-	deviceID := device.GetDeviceID()
-	log.Init(utils.Log_file_dic)       // 初始化日志目录
-	log.Info("Device id: " + deviceID) // 第一次启动记录
+	banner := `
+==========================================================================
+ _______  _______  _       __________________ _        _______  _       
+(  ____ \(  ____ \( (    /|\__   __/\__   __/( (    /|(  ____ \( \      
+| (_____ | (__    |   \ | |   | |      | |   |   \ | || (__    | |      
+(_____  )|  __)   | (\ \) |   | |      | |   | (\ \) ||  __)   | |      
+/\____) || (____/\| )  \  |   | |   ___) (___| )  \  || (____/\| (____/\
+\_______)(_______/|/    )_)   )_(   \_______/|/    )_)(_______/(_______/
+==========================================================================
+`

-	broker := fmt.Sprintf("tls://%s:%d", utils.MQTT_HOST, utils.MQTT_PORT)
-	username := deviceID
-	password := utils.PASSWORD
+	fmt.Println(banner)
+	deviceID := device.GetDeviceID()
+	log.Init(config.Log_file_dic)            // 初始化日志目录
+	log.Info("Device id: " + deviceID)       // 第一次启动记录
+	log.Println("版本号: ", config.APP_VERSION) // 第一次启动记录

 	var mqttSvc *MQTTService
 	firstFail := true // 标记是否第一次失败
 	for {
-		mqttSvc = NewMQTTService(broker, username, username, password, 60)
+		mqttSvc = NewMQTTService(config.MQTT_BROKER, deviceID, deviceID, config.PASSWORD, 60)
 		err := mqttSvc.Connect()
 		if err != nil {
 			if firstFail {
-				log.Error("物联网服务连接失败，请先注册设备. DeviceID: " + deviceID + " ")
+				log.Error("物联网服务连接失败，如未注册设备，请先注册: " + deviceID)
 				firstFail = false
 			}
-			time.Sleep(5 * time.Second) // 5秒后重试
+			time.Sleep(3 * time.Second) // 5秒后重试
 			continue
 		}
-		log.Info("物联网服务已启动")
 		break
 	}
 	defer mqttSvc.Close()

-	biz := NewBusinessService(mqttSvc, utils.PROJECT, utils.DEVICE_TPYE, deviceID)
+	biz := NewBusinessService(mqttSvc, deviceID)
 	for {
 		// MQTT业务
 		err := biz.Start()
@@ -47,7 +55,7 @@ func main() {
 			continue
 		}
 		// 个人业务
-		test()
+		//test()
 		break
 	}