1d5aff2f2e
1. 實作 MqttService 統一管理 MQTT 指令發送。 2. 遷移「遠端庫存與效期中心」至 MQTT 流程，並加入 batch_no (批號) 支援。 3. 修改 RemoteController 指令中心所有指令同步改由 MQTT 推送。 4. 修正 Go Gateway redis_consumer.go 確保 JSON 格式符合最新規格（移除寫死的 target/message_id）。 5. 實作 ProcessCommandAck 自動更新指令結果、異常回滾 (Rollback) 與機台自動亮燈 (Online)。 6. 完善 mqtt-communication-specs 規格文件，補齊所有指令定義與 ACK 範例。 7. 強化指令狀態保護，確保只有 pending 狀態的指令能被更新結果，防止重複處理。
67 lines
1.7 KiB
Go
67 lines
1.7 KiB
Go
package bridge
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"log"
|
||
|
||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
type RedisConsumer struct {
|
||
rdb *redis.Client
|
||
mqttClient mqtt.Client
|
||
queueKey string
|
||
}
|
||
|
||
func NewRedisConsumer(rdb *redis.Client, mqttClient mqtt.Client, queueKey string) *RedisConsumer {
|
||
return &RedisConsumer{rdb: rdb, mqttClient: mqttClient, queueKey: queueKey}
|
||
}
|
||
|
||
func (c *RedisConsumer) Start(ctx context.Context) {
|
||
log.Printf("Redis Consumer started, listening on queue: %s", c.queueKey)
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
default:
|
||
// 使用 BLPOP 阻塞式讀取，逾時設為 0 表示永久等待
|
||
result, err := c.rdb.BLPop(ctx, 0, c.queueKey).Result()
|
||
if err != nil {
|
||
if err != context.Canceled {
|
||
log.Printf("Redis BLPop error: %v", err)
|
||
}
|
||
continue
|
||
}
|
||
|
||
// result[0] 是 key，result[1] 是 value
|
||
var cmd map[string]interface{}
|
||
err = json.Unmarshal([]byte(result[1]), &cmd)
|
||
if err != nil {
|
||
log.Printf("Failed to unmarshal command: %v", err)
|
||
continue
|
||
}
|
||
|
||
serialNo, ok := cmd["serial_no"].(string)
|
||
if !ok || serialNo == "" {
|
||
log.Printf("Invalid or missing serial_no in outgoing command: %v", string(result[1]))
|
||
continue
|
||
}
|
||
|
||
// 將發送方才需要的欄位移除，保持 Payload 乾淨
|
||
delete(cmd, "serial_no")
|
||
delete(cmd, "timestamp")
|
||
|
||
topic := "machine/" + serialNo + "/command"
|
||
payload, _ := json.Marshal(cmd)
|
||
|
||
token := c.mqttClient.Publish(topic, 1, false, payload)
|
||
token.Wait()
|
||
|
||
commandType, _ := cmd["command"].(string)
|
||
log.Printf("Sent command [%s] to machine [%s] via topic: %s", commandType, serialNo, topic)
|
||
}
|
||
}
|
||
}
|