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

package mqtt

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
	"github.com/google/uuid"
	"github.com/sagikazarmark/slog-shim"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/reflect/protoreflect"
	"joylink.club/bj-rtsts-server/ts/protos/state"
)

var mqttClient *MqttClient
var clientId string

// 客户端
type MqttClient struct {
	cmc *MqttOptions
	cc  *autopaho.ClientConfig
	cm  *autopaho.ConnectionManager
}

// 初始化并启动MQTT客户端服务
func Startup(cmc *MqttOptions) error {
	initClientId()
	cmc.ClientId = clientId
	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{cmc: cmc, 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
}

// 初始化MQTT客户端id
func initClientId() {
	if clientId == "" {
		us := uuid.New().String()
		usl := len(us)
		sufix5 := us[usl-5 : usl]
		clientId = fmt.Sprintf("%s%d", sufix5, rand.New(rand.NewSource(time.Now().UnixNano())).Int()%1000)
	}
}

// 获取MQTT客户端id
func GetClientId() string {
	return clientId
}

// 发布数据
func 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 = mqttClient.cm.Publish(context.Background(), &paho.Publish{
		Topic:   topic,
		QoS:     0,
		Payload: b,
	})
	return err
}

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

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

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

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

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