32bc13ac1b
1. 補齊 ListenMqttQueue 中的 Job 類別引用，修復 mqtt-worker 崩潰重啟問題。
2. 實作 Go Gateway 的 Client ID 解析邏輯，正確將 SC_{serial}_{random} 格式還原為機台編號。
3. 修復 系統主題解析與 Laravel ProcessStatus Job 對齊，實現毫秒級連線狀態同步。
4. 在 ProcessStatus Job 中增加第二層 serial_no 清理保險，提高系統容錯。
5. 更新機台列表 UI 狀態顯示邏輯，支援 在線/斷線/故障 三態顯示與多語系。
6. 更新 MQTT 通訊規範 (Skill) 文件，確保全域開發定義的一致性。
7. 同步更新 zh_TW.json, en.json, ja.json 翻譯文件。
101 lines
2.9 KiB
Go
101 lines
2.9 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"star-cloud-gateway/config"
|
|
"star-cloud-gateway/internal/bridge"
|
|
"star-cloud-gateway/internal/handler"
|
|
"syscall"
|
|
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
func main() {
|
|
cfg := config.LoadConfig()
|
|
|
|
var rdb *redis.Client
|
|
var client mqtt.Client
|
|
|
|
// 1. 初始化 Redis (帶重試機制)
|
|
for {
|
|
rdb = redis.NewClient(&redis.Options{
|
|
Addr: cfg.RedisAddr,
|
|
Password: cfg.RedisPassword,
|
|
DB: 0,
|
|
})
|
|
if err := rdb.Ping(context.Background()).Err(); err != nil {
|
|
log.Printf("Waiting for Redis at %s... (%v)", cfg.RedisAddr, err)
|
|
time.Sleep(5 * time.Second)
|
|
continue
|
|
}
|
|
log.Printf("Connected to Redis at %s", cfg.RedisAddr)
|
|
break
|
|
}
|
|
|
|
// 加入 Laravel 前綴，確保雙方桶子同一個
|
|
prefixedIncomingKey := cfg.RedisPrefix + cfg.IncomingQueueKey
|
|
pusher := bridge.NewRedisPusher(rdb, prefixedIncomingKey)
|
|
mqttHandler := handler.NewMqttHandler(pusher)
|
|
|
|
// 2. 初始化 MQTT (帶重試機制)
|
|
opts := mqtt.NewClientOptions()
|
|
opts.AddBroker(cfg.MQTTAddr)
|
|
opts.SetClientID(cfg.MQTTClientID)
|
|
opts.SetUsername("star-cloud-gateway")
|
|
opts.SetPassword("StarCloudSecret999") // 與 Redis 中的 hash 對應
|
|
opts.SetCleanSession(false)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetMaxReconnectInterval(10 * time.Second)
|
|
|
|
opts.SetOnConnectHandler(func(c mqtt.Client) {
|
|
log.Printf("Connected to MQTT Broker as Gateway [star-cloud-gateway] at %s", cfg.MQTTAddr)
|
|
// 訂閱所有機台的上行 Topic
|
|
token := c.Subscribe("machine/+/+", 1, mqttHandler.HandleMessage)
|
|
token.Wait()
|
|
|
|
// 訂閱 EMQX 5 系統主題 (連線與斷線事件)
|
|
sysToken := c.Subscribe("$SYS/brokers/+/clients/+/connected", 1, mqttHandler.HandleMessage)
|
|
sysToken.Wait()
|
|
sysToken2 := c.Subscribe("$SYS/brokers/+/clients/+/disconnected", 1, mqttHandler.HandleMessage)
|
|
sysToken2.Wait()
|
|
|
|
if token.Error() != nil || sysToken.Error() != nil || sysToken2.Error() != nil {
|
|
log.Printf("Failed to subscribe to MQTT topics: %v %v %v", token.Error(), sysToken.Error(), sysToken2.Error())
|
|
} else {
|
|
log.Println("Successfully subscribed to machine/+/+ and $SYS topics")
|
|
}
|
|
})
|
|
|
|
client = mqtt.NewClient(opts)
|
|
for {
|
|
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
|
log.Printf("Waiting for MQTT Broker at %s... (%v)", cfg.MQTTAddr, token.Error())
|
|
time.Sleep(5 * time.Second)
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
|
|
// 3. 啟動 Redis Consumer (下行指令)
|
|
prefixedOutgoingKey := cfg.RedisPrefix + cfg.OutgoingQueueKey
|
|
consumer := bridge.NewRedisConsumer(rdb, client, prefixedOutgoingKey)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
go consumer.Start(ctx)
|
|
|
|
// 4. 阻塞並監聽信號
|
|
sigChan := make(chan os.Signal, 1)
|
|
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
|
|
<-sigChan
|
|
|
|
log.Println("Shutting down Gateway...")
|
|
cancel() // 停止 Consumer
|
|
client.Disconnect(250)
|
|
rdb.Close()
|
|
}
|