rts-sim-testing-service/mqtt/client.go

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/message_server/ms_api"
	"joylink.club/bj-rtsts-server/ts/protos/state"
)

var mqttClient *MqttClient

// 客户端
type MqttClient struct {
	cc    *autopaho.ClientConfig
	cm    *autopaho.ConnectionManager
	tasks map[string][]ms_api.MsgTask
}

// 初始化并启动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, tasks: make(map[string][]ms_api.MsgTask)}
	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)
	}
	_, err = client.cm.Publish(context.Background(), &paho.Publish{
		Topic:   topic,
		QoS:     0,
		Payload: b,
	})
	return err
}

// 发布任务
func (client *MqttClient) PublishTask(simulationId string, tasks ...ms_api.MsgTask) {
	client.tasks[simulationId] = append(client.tasks[simulationId], tasks...)
}

// 停止任务
func (client *MqttClient) CloseTask(simulationId string) {
	tasks, ok := client.tasks[simulationId]
	if !ok {
		return
	}
	for _, task := range tasks {
		task.Stop()
	}
	client.tasks[simulationId] = nil
}

// 发送仿真状态数据
func (client *MqttClient) PubSimulationState(simulationId string, msg *state.SimulationStatus) error {
	return client.pub(GetStateTopic(simulationId), msg)
}

// 发送IBP状态数据
func (client *MqttClient) PubIBPState(simulationId string, mapId int32, stationId uint32, msg *state.PushedDevicesStatus) error {
	return client.pub(GetIbpTopic(simulationId, mapId, stationId), msg)
}

// 发送PSL状态数据
func (client *MqttClient) PubPSLState(simulationId string, mapId int32, boxId uint32, msg *state.PushedDevicesStatus) error {
	return client.pub(GetPslTopic(simulationId, mapId, boxId), msg)
}

// 发送继电器状态数据
func (client *MqttClient) PubRCCState(simulationId string, mapId int32, msg *state.PushedDevicesStatus) error {
	return client.pub(GetRccTopic(simulationId, mapId), msg)
}

// 发送站场图状态数据
func (client *MqttClient) PubSfpState(simulationId string, mapId int32, msg *state.PushedDevicesStatus) error {
	return client.pub(GetSfpTopic(simulationId, mapId), msg)
}