package mqtt import ( "context" "fmt" "github.com/eclipse/paho.golang/autopaho" "github.com/eclipse/paho.golang/paho" "github.com/sagikazarmark/slog-shim" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/reflect/protoreflect" "joylink.club/bj-rtsts-server/dto/state_proto" ) var mqttClient *MqttClient // 客户端 type MqttClient struct { cc *autopaho.ClientConfig cm *autopaho.ConnectionManager } // 初始化并启动MQTT客户端服务 func Startup(cmc *MqttOptions) error { if err := checkConfig(cmc); err != nil { return err } cc, err := cmc.tryInto() if err != nil { return err } cm, err := autopaho.NewConnection(context.Background(), *cc) if err != nil { return err } mqttClient = &MqttClient{cc: cc, cm: cm} return nil } // 检查配置信息 func checkConfig(cmc *MqttOptions) error { if cmc.AppId == "" { return fmt.Errorf("应用编号不能为空") } if cmc.ClientId == "" { return fmt.Errorf("客户端编号不能为空") } if cmc.BrokerUrl == "" { return fmt.Errorf("MQTT代理服务地址不能为空") } if cmc.Username == "" { return fmt.Errorf("MQTT用户名不能为空") } if cmc.Password == "" { return fmt.Errorf("MQTT密码不能为空") } return nil } func GetMsgClient() *MqttClient { return mqttClient } // 获取MQTT客户端id func GetClientId() string { return mqttClient.cc.ClientConfig.ClientID } // 发布数据 func (client *MqttClient) pub(topic string, data protoreflect.ProtoMessage) error { if data == nil { return fmt.Errorf("发布数据引用为nil") } b, err := proto.Marshal(data) if err != nil { return err } if !MatchTopic(topic) { slog.Error("未知发布主题", "topic", topic, "data", data) return fmt.Errorf("未知发布主题: topic=%s", topic) } // else { // slog.Debug("发布主题", "topic", topic, "data", data) // } _, err = client.cm.Publish(context.Background(), &paho.Publish{ Topic: topic, QoS: 0, Payload: b, }) return err } // 发送仿真状态数据 func (client *MqttClient) PubSimulationState(simulationId string, msg *state_proto.SimulationStatus) error { return client.pub(GetStateTopic(simulationId), msg) } func (c *MqttClient) PubTpapiServiceState(simulationId string, msg *state_proto.SimulationThirdPartyApiService) error { return c.pub(GetTpapiServiceTopic(simulationId), msg) } // 发送IBP状态数据 func (client *MqttClient) PubIBPState(simulationId string, mapId int32, ibpId uint32, msg *state_proto.PushedDevicesStatus) error { return client.pub(GetIbpTopic(simulationId, mapId, ibpId), msg) } // 发送PSL状态数据 func (client *MqttClient) PubPSLState(simulationId string, mapId int32, boxId uint32, msg *state_proto.PushedDevicesStatus) error { return client.pub(GetPslTopic(simulationId, mapId, boxId), msg) } // 发送继电器状态数据 func (client *MqttClient) PubRCCState(simulationId string, mapId int32, msg *state_proto.PushedDevicesStatus) error { return client.pub(GetRccTopic(simulationId, mapId), msg) } // 发送站场图状态数据 func (client *MqttClient) PubSfpState(simulationId string, mapId int32, msg *state_proto.PushedDevicesStatus) error { return client.pub(GetSfpTopic(simulationId, mapId), msg) } // 发送列车控制状态 func (client *MqttClient) PubTrainControlState(simulationId string, trainId string, msg *state_proto.TrainControlState) error { return client.pub(GetTrainControlTopic(simulationId, trainId), msg) }