130 lines
3.6 KiB
Go
130 lines
3.6 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"errors"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/yutongmqtt"
|
|
)
|
|
|
|
type MQTTClientConfig struct {
|
|
EndpointName string
|
|
Broker string
|
|
ClientID string
|
|
Username string
|
|
Password string
|
|
Topics []string
|
|
QoS byte
|
|
Sink eventbus.Sink
|
|
Logger *slog.Logger
|
|
}
|
|
|
|
type MQTTClient struct {
|
|
cfg MQTTClientConfig
|
|
client mqtt.Client
|
|
}
|
|
|
|
func NewMQTTClient(cfg MQTTClientConfig) (*MQTTClient, error) {
|
|
if strings.TrimSpace(cfg.Broker) == "" {
|
|
return nil, errors.New("mqtt broker is required")
|
|
}
|
|
if strings.TrimSpace(cfg.ClientID) == "" {
|
|
return nil, errors.New("mqtt client id is required")
|
|
}
|
|
if len(cfg.Topics) == 0 {
|
|
return nil, errors.New("mqtt topics are required")
|
|
}
|
|
if cfg.Sink == nil {
|
|
return nil, errors.New("sink is required")
|
|
}
|
|
if cfg.Logger == nil {
|
|
cfg.Logger = slog.Default()
|
|
}
|
|
if cfg.EndpointName == "" {
|
|
cfg.EndpointName = "yutong"
|
|
}
|
|
return &MQTTClient{cfg: cfg}, nil
|
|
}
|
|
|
|
func (c *MQTTClient) Start(ctx context.Context) error {
|
|
opts := mqtt.NewClientOptions().
|
|
AddBroker(c.cfg.Broker).
|
|
SetClientID(c.cfg.ClientID).
|
|
SetUsername(c.cfg.Username).
|
|
SetPassword(c.cfg.Password).
|
|
SetCleanSession(false).
|
|
SetAutoReconnect(true).
|
|
SetConnectRetry(true).
|
|
SetConnectRetryInterval(5 * time.Second).
|
|
SetKeepAlive(20 * time.Second).
|
|
SetConnectTimeout(10 * time.Second)
|
|
|
|
opts.SetDefaultPublishHandler(func(_ mqtt.Client, message mqtt.Message) {
|
|
c.handleMessage(ctx, message.Topic(), message.Payload())
|
|
})
|
|
opts.OnConnect = func(client mqtt.Client) {
|
|
for _, topic := range c.cfg.Topics {
|
|
token := client.Subscribe(topic, c.cfg.QoS, nil)
|
|
if token.Wait() && token.Error() != nil {
|
|
c.cfg.Logger.Error("mqtt subscribe failed", "broker", c.cfg.Broker, "topic", topic, "error", token.Error())
|
|
continue
|
|
}
|
|
c.cfg.Logger.Info("mqtt subscribed", "broker", c.cfg.Broker, "topic", topic, "qos", c.cfg.QoS)
|
|
}
|
|
}
|
|
opts.OnConnectionLost = func(_ mqtt.Client, err error) {
|
|
c.cfg.Logger.Warn("mqtt connection lost", "broker", c.cfg.Broker, "error", err)
|
|
}
|
|
|
|
c.client = mqtt.NewClient(opts)
|
|
token := c.client.Connect()
|
|
if token.Wait() && token.Error() != nil {
|
|
return token.Error()
|
|
}
|
|
go func() {
|
|
<-ctx.Done()
|
|
if c.client != nil && c.client.IsConnected() {
|
|
c.client.Disconnect(250)
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) {
|
|
receivedAtMS := time.Now().UnixMilli()
|
|
env, err := yutongmqtt.ParseMessage(c.cfg.EndpointName, topic, payload, receivedAtMS)
|
|
if err != nil {
|
|
env = envelope.FrameEnvelope{
|
|
Protocol: envelope.ProtocolYutongMQTT,
|
|
MessageID: "MQTT",
|
|
VehicleKeyHint: "unknown",
|
|
SourceEndpoint: "mqtt://" + c.cfg.EndpointName + topic,
|
|
EventTimeMS: receivedAtMS,
|
|
ReceivedAtMS: receivedAtMS,
|
|
RawText: string(payload),
|
|
RawHex: strings.ToUpper(hex.EncodeToString(payload)),
|
|
ParseStatus: envelope.ParseBadFrame,
|
|
ParseError: err.Error(),
|
|
}
|
|
env.EventID = env.StableEventID()
|
|
}
|
|
if err := c.cfg.Sink.PublishRaw(ctx, env); err != nil {
|
|
c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
|
|
return
|
|
}
|
|
if env.ParseStatus == envelope.ParseBadFrame {
|
|
return
|
|
}
|
|
if err := c.cfg.Sink.PublishUnified(ctx, env); err != nil {
|
|
c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
|
|
}
|
|
}
|