diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index cb653141..1069fb1b 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -68,6 +68,28 @@ func main() { } }() } + if envBool("YUTONG_MQTT_ENABLED", false) { + client, err := gateway.NewMQTTClient(gateway.MQTTClientConfig{ + EndpointName: env("YUTONG_MQTT_ENDPOINT", "yutong"), + Broker: env("YUTONG_MQTT_URI", ""), + ClientID: env("YUTONG_MQTT_CLIENT_ID", "lingniu-go-yutong-mqtt"), + Username: env("YUTONG_MQTT_USERNAME", ""), + Password: env("YUTONG_MQTT_PASSWORD", ""), + Topics: splitCSV(env("YUTONG_MQTT_TOPICS", env("YUTONG_MQTT_TOPIC", "/ytforward/shln/+"))), + QoS: byte(envInt("YUTONG_MQTT_QOS", 2)), + Sink: sink, + Logger: logger, + }) + if err != nil { + logger.Error("build yutong mqtt client failed", "error", err) + os.Exit(1) + } + if err := client.Start(ctx); err != nil { + logger.Error("start yutong mqtt client failed", "error", err) + os.Exit(1) + } + logger.Info("yutong mqtt client started") + } logger.Info("vehicle gateway started") select { @@ -122,6 +144,14 @@ func envInt(key string, fallback int) int { return out } +func envBool(key string, fallback bool) bool { + value := strings.ToLower(strings.TrimSpace(os.Getenv(key))) + if value == "" { + return fallback + } + return value == "1" || value == "true" || value == "yes" || value == "on" +} + func splitCSV(value string) []string { var out []string for _, item := range strings.Split(value, ",") { diff --git a/go/vehicle-gateway/go.mod b/go/vehicle-gateway/go.mod index 9008363a..022aa215 100644 --- a/go/vehicle-gateway/go.mod +++ b/go/vehicle-gateway/go.mod @@ -4,6 +4,7 @@ go 1.26 require ( github.com/alicebob/miniredis/v2 v2.35.0 + github.com/eclipse/paho.mqtt.golang v1.5.1 github.com/go-sql-driver/mysql v1.9.3 github.com/redis/go-redis/v9 v9.17.2 github.com/segmentio/kafka-go v0.4.49 @@ -15,11 +16,13 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/google/uuid v1.6.0 // indirect - github.com/gorilla/websocket v1.5.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.15.9 // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/yuin/gopher-lua v1.1.1 // indirect + golang.org/x/net v0.44.0 // indirect + golang.org/x/sync v0.17.0 // indirect ) diff --git a/go/vehicle-gateway/go.sum b/go/vehicle-gateway/go.sum index e246a9d5..00fee118 100644 --- a/go/vehicle-gateway/go.sum +++ b/go/vehicle-gateway/go.sum @@ -13,13 +13,16 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= +github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY= @@ -55,10 +58,12 @@ github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= -golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8= -golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8= -golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= -golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= +golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= +golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go new file mode 100644 index 00000000..9403ffa3 --- /dev/null +++ b/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) + } +} diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go new file mode 100644 index 00000000..3d92da11 --- /dev/null +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -0,0 +1,61 @@ +package gateway + +import ( + "context" + "log/slog" + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) { + sink := &recordingSink{} + client, err := NewMQTTClient(MQTTClientConfig{ + EndpointName: "endpoint-a", + Broker: "tcp://127.0.0.1:1883", + ClientID: "test-client", + Topics: []string{"/ytforward/shln/+"}, + Sink: sink, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + }) + if err != nil { + t.Fatalf("NewMQTTClient() error = %v", err) + } + + client.handleMessage(context.Background(), "/ytforward/shln/dev1", []byte(`{ + "device":"LTEST000000000001", + "time":"20260413100000", + "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7} + }`)) + + if len(sink.raw) != 1 || len(sink.unified) != 1 { + t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) + } + if sink.raw[0].Protocol != envelope.ProtocolYutongMQTT || sink.raw[0].VIN != "LTEST000000000001" { + t.Fatalf("unexpected raw envelope: %#v", sink.raw[0]) + } +} + +func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { + sink := &recordingSink{} + client, err := NewMQTTClient(MQTTClientConfig{ + EndpointName: "endpoint-a", + Broker: "tcp://127.0.0.1:1883", + ClientID: "test-client", + Topics: []string{"/ytforward/shln/+"}, + Sink: sink, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + }) + if err != nil { + t.Fatalf("NewMQTTClient() error = %v", err) + } + + client.handleMessage(context.Background(), "/ytforward/shln/bad", []byte("{bad-json")) + + if len(sink.raw) != 1 || len(sink.unified) != 0 { + t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) + } + if sink.raw[0].ParseStatus != envelope.ParseBadFrame { + t.Fatalf("parse status = %q", sink.raw[0].ParseStatus) + } +} diff --git a/go/vehicle-gateway/internal/protocol/yutongmqtt/parser.go b/go/vehicle-gateway/internal/protocol/yutongmqtt/parser.go new file mode 100644 index 00000000..9eff6fb5 --- /dev/null +++ b/go/vehicle-gateway/internal/protocol/yutongmqtt/parser.go @@ -0,0 +1,189 @@ +package yutongmqtt + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +var ErrInvalidJSON = errors.New("yutong mqtt invalid json") + +func ParseMessage(endpoint string, topic string, payload []byte, receivedAtMS int64) (envelope.FrameEnvelope, error) { + root := map[string]any{} + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.UseNumber() + if err := decoder.Decode(&root); err != nil { + return envelope.FrameEnvelope{}, fmt.Errorf("%w: %v", ErrInvalidJSON, err) + } + + data := objectValue(root, "data") + if data == nil { + data = root + } + deviceID := firstText(root, "device", "deviceId", "terminalId", "terminalID", "imei", "IMEI", "mqttDeviceId") + externalVIN := firstText(root, "vin", "VIN", "externalVin", "vehicleVin") + vin := externalVIN + if vin == "" && looksLikeVIN(deviceID) { + vin = deviceID + } + plate := firstText(root, "plateNo", "carNo", "plate", "vehicleNo") + phone := firstText(root, "sim", "phone", "mobile") + eventTimeMS := parseTimeMS(firstText(root, "time", "deviceTime", "eventTime", "gpsTime"), receivedAtMS) + fields := fieldsFromData(data) + + parsed := map[string]any{ + "endpoint": endpoint, + "topic": topic, + "root": root, + "data": data, + } + env := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + MessageID: "MQTT", + VIN: vin, + VehicleKeyHint: deviceID, + Phone: phone, + DeviceID: deviceID, + Plate: plate, + SourceEndpoint: "mqtt://" + endpoint + topic, + EventTimeMS: eventTimeMS, + ReceivedAtMS: receivedAtMS, + RawText: string(payload), + Parsed: parsed, + Fields: fields, + ParseStatus: envelope.ParseOK, + } + env.EventID = env.StableEventID() + return env, nil +} + +func fieldsFromData(data map[string]any) map[string]any { + fields := map[string]any{} + if speed, ok := firstFloat(data, "METER_SPEED", "speed", "speed_kmh"); ok { + fields[envelope.FieldSpeedKMH] = speed + } + if mileage, ok := firstFloat(data, "TOTAL_MILEAGE", "totalMileage", "total_mileage_km"); ok { + fields[envelope.FieldTotalMileageKM] = mileage + } + if soc, ok := firstFloat(data, "BATTERY_CAPACITY_SOC", "soc", "SOC", "soc_percent"); ok { + fields[envelope.FieldSOCPercent] = soc + } + if longitude, ok := firstCoord(data, "LONGITUDE", "longitude", "lng"); ok { + fields[envelope.FieldLongitude] = longitude + } + if latitude, ok := firstCoord(data, "LATITUDE", "latitude", "lat"); ok { + fields[envelope.FieldLatitude] = latitude + } + if direction, ok := firstFloat(data, "GPSDirection", "direction", "direction_deg"); ok { + fields["direction_deg"] = direction + } + if gear, ok := firstFloat(data, "ON_GEAR", "gear"); ok { + fields["gear"] = gear + } + return fields +} + +func objectValue(root map[string]any, key string) map[string]any { + value, ok := root[key] + if !ok { + return nil + } + typed, ok := value.(map[string]any) + if !ok { + return nil + } + return typed +} + +func firstText(root map[string]any, keys ...string) string { + for _, key := range keys { + if value, ok := root[key]; ok && value != nil { + text := strings.TrimSpace(fmt.Sprint(value)) + if text != "" && text != "" { + return text + } + } + } + return "" +} + +func firstFloat(root map[string]any, keys ...string) (float64, bool) { + for _, key := range keys { + value, ok := root[key] + if !ok || value == nil { + continue + } + if parsed, ok := toFloat(value); ok { + return parsed, true + } + } + return 0, false +} + +func firstCoord(root map[string]any, keys ...string) (float64, bool) { + value, ok := firstFloat(root, keys...) + if !ok { + return 0, false + } + if value == 0 || int64(value) == 999999 { + return 0, false + } + if value > 1000 || value < -1000 { + value = value / 1_000_000 + } + return value, true +} + +func toFloat(value any) (float64, bool) { + switch typed := value.(type) { + case json.Number: + parsed, err := typed.Float64() + return parsed, err == nil + case float64: + return typed, true + case float32: + return float64(typed), true + case int: + return float64(typed), true + case int64: + return float64(typed), true + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) + return parsed, err == nil + default: + return 0, false + } +} + +func parseTimeMS(raw string, fallback int64) int64 { + raw = strings.TrimSpace(raw) + if raw == "" { + return fallback + } + if len(raw) == 14 { + if parsed, err := time.ParseInLocation("20060102150405", raw, time.FixedZone("Asia/Shanghai", 8*3600)); err == nil { + return parsed.UnixMilli() + } + } + if parsed, err := time.Parse(time.RFC3339, raw); err == nil { + return parsed.UnixMilli() + } + if epoch, err := strconv.ParseInt(raw, 10, 64); err == nil { + if epoch > 10_000_000_000 { + return epoch + } + return epoch * 1000 + } + return fallback +} + +func looksLikeVIN(value string) bool { + value = strings.TrimSpace(value) + return len(value) == 17 +} diff --git a/go/vehicle-gateway/internal/protocol/yutongmqtt/parser_test.go b/go/vehicle-gateway/internal/protocol/yutongmqtt/parser_test.go new file mode 100644 index 00000000..f348322c --- /dev/null +++ b/go/vehicle-gateway/internal/protocol/yutongmqtt/parser_test.go @@ -0,0 +1,83 @@ +package yutongmqtt + +import ( + "errors" + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestParseMessageMapsYutongPayloadToEnvelope(t *testing.T) { + payload := []byte(`{ + "device": "LTEST000000000001", + "time": "20260413100000", + "plateNo": "豫A12345", + "data": { + "METER_SPEED": 52.3, + "TOTAL_MILEAGE": 123456.7, + "BATTERY_CAPACITY_SOC": 70, + "LONGITUDE": 116397128, + "LATITUDE": 39916527, + "GPSDirection": 88 + } + }`) + + env, err := ParseMessage("endpoint-a", "/ytforward/shln/dev1", payload, 1782745114999) + if err != nil { + t.Fatalf("ParseMessage() error = %v", err) + } + if env.Protocol != envelope.ProtocolYutongMQTT { + t.Fatalf("protocol = %q", env.Protocol) + } + if env.VIN != "LTEST000000000001" || env.DeviceID != "LTEST000000000001" { + t.Fatalf("identity fields = vin:%q device:%q", env.VIN, env.DeviceID) + } + if env.Plate != "豫A12345" { + t.Fatalf("plate = %q", env.Plate) + } + assertFloatField(t, env, envelope.FieldSpeedKMH, 52.3) + assertFloatField(t, env, envelope.FieldTotalMileageKM, 123456.7) + assertFloatField(t, env, envelope.FieldSOCPercent, 70) + assertFloatField(t, env, envelope.FieldLongitude, 116.397128) + assertFloatField(t, env, envelope.FieldLatitude, 39.916527) + assertFloatField(t, env, "direction_deg", 88) + if env.EventTimeMS == 1782745114999 { + t.Fatal("event time should come from device time") + } + if env.RawText == "" || env.Parsed["data"] == nil { + t.Fatalf("raw/parsed missing: %#v", env) + } +} + +func TestParseMessageUsesDeviceAsVehicleKeyWhenNotVIN(t *testing.T) { + payload := []byte(`{"device":"YT-DEVICE-001","time":"20260413100000","data":{"METER_SPEED":"12.3"}}`) + env, err := ParseMessage("endpoint-a", "/ytforward/shln/YT-DEVICE-001", payload, 1782745114999) + if err != nil { + t.Fatalf("ParseMessage() error = %v", err) + } + if env.VIN != "" { + t.Fatalf("non-vin device should not be treated as vin: %q", env.VIN) + } + if env.VehicleKey() != "YT-DEVICE-001" { + t.Fatalf("vehicle key = %q", env.VehicleKey()) + } + assertFloatField(t, env, envelope.FieldSpeedKMH, 12.3) +} + +func TestParseMessageRejectsMalformedJSON(t *testing.T) { + _, err := ParseMessage("endpoint-a", "/bad", []byte("{bad-json"), 1782745114999) + if !errors.Is(err, ErrInvalidJSON) { + t.Fatalf("error = %v", err) + } +} + +func assertFloatField(t *testing.T, env envelope.FrameEnvelope, key string, want float64) { + t.Helper() + got, ok := env.Fields[key].(float64) + if !ok { + t.Fatalf("field %s missing or not float64: %#v", key, env.Fields[key]) + } + if got != want { + t.Fatalf("field %s = %v, want %v", key, got, want) + } +}