diff --git a/.gitea/workflows/deploy-demo.yaml b/.gitea/workflows/deploy-demo.yaml index 7f57b12..48f80cc 100644 --- a/.gitea/workflows/deploy-demo.yaml +++ b/.gitea/workflows/deploy-demo.yaml @@ -38,6 +38,8 @@ jobs: - name: Step 1 - Build & Deploy Containers run: | WWWGROUP=1000 WWWUSER=1000 docker compose -f compose.yaml up -d --build --wait + # 使用熱重載套用 ACL 變更,不中斷現有連線 + docker exec star-cloud-emqx emqx_ctl authorization reload - name: Step 2 - Deploy Code run: | @@ -75,5 +77,10 @@ jobs: " docker exec star-cloud-laravel chmod -R 775 /var/www/html/storage /var/www/html/bootstrap/cache - - name: Step 4 - Cleanup + - name: Step 4 - Restart Long-running Workers + run: | + # 重啟 mqtt:listen 常駐行程以載入最新 PHP 程式碼 + docker restart star-cloud-mqtt-worker + + - name: Step 5 - Cleanup run: docker image prune -f diff --git a/docker/emqx/acl.conf b/docker/emqx/acl.conf index 128fc31..18f6e3a 100644 --- a/docker/emqx/acl.conf +++ b/docker/emqx/acl.conf @@ -1,6 +1,4 @@ -{allow, {user, "star-cloud-gateway"}, subscribe, ["$SYS/brokers/+/clients/+/connected", "$SYS/brokers/+/clients/+/disconnected"]}. {allow, {username, "dashboard"}, subscribe, ["$SYS/#"]}. {allow, {ipaddr, "127.0.0.1"}, all, ["$SYS/#", "#"]}. -{allow, {username, "star-cloud-gateway"}, subscribe, ["$SYS/#"]}. {deny, all, subscribe, ["$SYS/#", {eq, "#"}, {eq, "+/#"}]}. {allow, all}. diff --git a/mqtt-gateway/internal/handler/mqtt_handler.go b/mqtt-gateway/internal/handler/mqtt_handler.go index cef03c4..792d92f 100644 --- a/mqtt-gateway/internal/handler/mqtt_handler.go +++ b/mqtt-gateway/internal/handler/mqtt_handler.go @@ -23,96 +23,34 @@ func (h *MqttHandler) HandleMessage(client mqtt.Client, msg mqtt.Message) { log.Printf("Received message on topic: %s", topic) - var serialNo, msgType string + // 業務主題格式: + // 3 層: machine/{serialNo}/{type} (如 heartbeat, error, transaction, status) + // 4 層: machine/{serialNo}/{type}/{subType} (如 command/ack) + parts := strings.Split(topic, "/") + if len(parts) < 3 { + return + } + + serialNo := parts[1] + msgType := parts[2] + + // 4 層 Topic:將 type/subType 組合為 type_subType (如 command/ack → command_ack) + if len(parts) >= 4 { + msgType = parts[2] + "_" + parts[3] + } + var jsonPayload interface{} - - // 檢查是否為系統主題 ($SYS) - if strings.HasPrefix(topic, "$SYS/") { - parts := strings.Split(topic, "/") - - // 尋找事件類型 (connected 或 disconnected) - var event string - for _, p := range parts { - if p == "connected" || p == "disconnected" { - event = p - break - } - } - - // 尋找 Client ID (通常位於 "clients" 的下一層) - for i, p := range parts { - if p == "clients" && i+1 < len(parts) { - serialNo = parts[i+1] - break - } - } - - if event != "" && serialNo != "" { - // 解析系統 JSON Payload (補強:有些版本會把 ClientID 放在 Payload 裡) - var sysData map[string]interface{} - if err := json.Unmarshal(payload, &sysData); err == nil { - if cid, ok := sysData["clientid"].(string); ok { - serialNo = cid - } - } - - msgType = "status" - statusValue := "online" - if event == "disconnected" { - statusValue = "offline" - } - jsonPayload = map[string]string{"status": statusValue} - - log.Printf("Detected system event: Client [%s] is now [%s]", serialNo, statusValue) - } else { - return - } - } else { - // 一般業務主題 - // 3 層: machine/{serialNo}/{type} (如 heartbeat, error, transaction) - // 4 層: machine/{serialNo}/{type}/{subType} (如 command/ack) - parts := strings.Split(topic, "/") - if len(parts) < 3 { - return - } - serialNo = parts[1] - msgType = parts[2] - - // 4 層 Topic:將 type/subType 組合為 type_subType (如 command/ack → command_ack) - if len(parts) >= 4 { - msgType = parts[2] + "_" + parts[3] - } - - err := json.Unmarshal(payload, &jsonPayload) - if err != nil { - // 如果不是有效的 JSON,則視為純文字/數值 (Raw Value) - jsonPayload = string(payload) - } + err := json.Unmarshal(payload, &jsonPayload) + if err != nil { + // 如果不是有效的 JSON,則視為純文字/數值 (Raw Value) + jsonPayload = string(payload) } bridgePayload := bridge.BridgePayload{ Type: msgType, - SerialNo: cleanSerialNo(serialNo), + SerialNo: serialNo, Payload: jsonPayload, } h.pusher.Push(bridgePayload) } - -// cleanSerialNo 處理 Client ID 格式,將 SC_M0408_pndk 轉為 M0408 -func cleanSerialNo(clientID string) string { - if clientID == "" { - return "" - } - - // 1. 移除已知前綴 SC_ - cleaned := strings.TrimPrefix(clientID, "SC_") - - // 2. 如果中間還有底線,通常代表隨機碼,只取前面部分 - // 例如 M0408_test -> M0408 - if idx := strings.Index(cleaned, "_"); idx != -1 { - return cleaned[:idx] - } - - return cleaned -} diff --git a/mqtt-gateway/main.go b/mqtt-gateway/main.go index 6e7d2f8..d6d4d35 100644 --- a/mqtt-gateway/main.go +++ b/mqtt-gateway/main.go @@ -63,16 +63,10 @@ func main() { ackToken := c.Subscribe("machine/+/command/ack", 1, mqttHandler.HandleMessage) ackToken.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 || ackToken.Error() != nil || sysToken.Error() != nil || sysToken2.Error() != nil { - log.Printf("Failed to subscribe to MQTT topics: %v %v %v %v", token.Error(), ackToken.Error(), sysToken.Error(), sysToken2.Error()) + if token.Error() != nil || ackToken.Error() != nil { + log.Printf("Failed to subscribe to MQTT topics: %v %v", token.Error(), ackToken.Error()) } else { - log.Println("Successfully subscribed to machine/+/+, machine/+/command/ack and $SYS topics") + log.Println("Successfully subscribed to machine/+/+ and machine/+/command/ack") } }) diff --git a/star-cloud-demo-day.pptx b/star-cloud-demo-day.pptx new file mode 100644 index 0000000..e701233 Binary files /dev/null and b/star-cloud-demo-day.pptx differ