feat: add yutong mqtt ingestion

go/vehicle-gateway/internal/gateway/mqtt_client.go

@@ -0,0 +1,129 @@
+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)
+	}
+}