bf43a33df3
1. 整合 MQTT Gateway 並支援動態 Redis Key 前綴隔離，解決多環境衝突。 2. 修正本地開發環境與 Demo 環境的埠號映射 (80/8080) 與 Vite HMR (5175) 設定。 3. 更新 CI/CD 部署流程，新增自動同步程式碼至 MQTT Worker 與 Queue 容器。 4. 擴充 .env.example 以提供完整的 MQTT/EMQX 相關變數範本。 5. 更新模擬腳本以符合現有測試資料。
94 lines
2.5 KiB
Go
94 lines
2.5 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()
|
|
if token.Error() != nil {
|
|
log.Printf("Failed to subscribe to machine/+/+: %v", token.Error())
|
|
} else {
|
|
log.Println("Successfully subscribed to machine/+/+")
|
|
}
|
|
})
|
|
|
|
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()
|
|
}
|