|
| 1 | +diff --git a/datastore/mqtt/mqtt_payload.go b/datastore/mqtt/mqtt_payload.go |
| 2 | +index 4c16039..38ac17e 100644 |
| 3 | +--- a/datastore/mqtt/mqtt_payload.go |
| 4 | ++++ b/datastore/mqtt/mqtt_payload.go |
| 5 | +@@ -11,12 +11,35 @@ import ( |
| 6 | + "google.golang.org/protobuf/reflect/protoreflect" |
| 7 | + ) |
| 8 | + |
| 9 | ++type vehicleFieldEnvelope struct { |
| 10 | ++ Value interface{} `json:"value"` |
| 11 | ++ TS string `json:"ts,omitempty"` |
| 12 | ++} |
| 13 | ++ |
| 14 | ++// marshalVehicleField preserves Payload.CreatedAt alongside each MQTT value. |
| 15 | ++// TeslaSync uses this source timestamp for ordering and keeps broker receipt |
| 16 | ++// time separately, so persistent-session replay cannot move old telemetry into |
| 17 | ++// the current drive or charging session. |
| 18 | ++// |
| 19 | ++// TeslaSync modification: the upstream producer publishes only the bare value |
| 20 | ++// and therefore discards CreatedAt at this transport boundary. |
| 21 | ++func marshalVehicleField(value interface{}, payload *protos.Payload) ([]byte, error) { |
| 22 | ++ envelope := vehicleFieldEnvelope{Value: value} |
| 23 | ++ if payload != nil { |
| 24 | ++ createdAt := payload.GetCreatedAt() |
| 25 | ++ if createdAt != nil && createdAt.CheckValid() == nil { |
| 26 | ++ envelope.TS = createdAt.AsTime().UTC().Format(time.RFC3339Nano) |
| 27 | ++ } |
| 28 | ++ } |
| 29 | ++ return json.Marshal(envelope) |
| 30 | ++} |
| 31 | ++ |
| 32 | + func (p *Producer) processVehicleFields(rec *telemetry.Record, payload *protos.Payload) ([]pahomqtt.Token, error) { |
| 33 | + var tokens []pahomqtt.Token |
| 34 | + convertedPayload := p.payloadToMap(payload) |
| 35 | + for key, value := range convertedPayload { |
| 36 | + mqttTopicName := fmt.Sprintf("%s/%s/v/%s", p.config.TopicBase, rec.Vin, key) |
| 37 | +- jsonValue, err := json.Marshal(value) |
| 38 | ++ jsonValue, err := marshalVehicleField(value, payload) |
| 39 | + if err != nil { |
| 40 | + return tokens, fmt.Errorf("failed to marshal JSON for MQTT topic %s: %v", mqttTopicName, err) |
| 41 | + } |
| 42 | +diff --git a/datastore/mqtt/mqtt_test.go b/datastore/mqtt/mqtt_test.go |
| 43 | +index a935516..d03f7d8 100644 |
| 44 | +--- a/datastore/mqtt/mqtt_test.go |
| 45 | ++++ b/datastore/mqtt/mqtt_test.go |
| 46 | +@@ -219,7 +219,7 @@ var _ = Describe("MQTTProducer", func() { |
| 47 | + }, |
| 48 | + }, |
| 49 | + }, |
| 50 | +- CreatedAt: timestamppb.Now(), |
| 51 | ++ CreatedAt: timestamppb.New(time.Date(2026, time.August, 20, 7, 8, 9, 123456789, time.UTC)), |
| 52 | + } |
| 53 | + |
| 54 | + payloadBytes, err := proto.Marshal(payload) |
| 55 | +@@ -248,10 +248,10 @@ var _ = Describe("MQTTProducer", func() { |
| 56 | + locationTopic := "test/topic/TEST123/v/Location" |
| 57 | + batteryLevelTopic := "test/topic/TEST123/v/BatteryLevel" |
| 58 | + |
| 59 | +- vehicleNameValue := "\"My Tesla\"" |
| 60 | +- invalidValue := "null" |
| 61 | +- locationValue := "{\"latitude\":37.7749,\"longitude\":-122.4194}" |
| 62 | +- batterLevelValue := "75.5" |
| 63 | ++ vehicleNameValue := `{"value":"My Tesla","ts":"2026-08-20T07:08:09.123456789Z"}` |
| 64 | ++ invalidValue := `{"value":null,"ts":"2026-08-20T07:08:09.123456789Z"}` |
| 65 | ++ locationValue := `{"value":{"latitude":37.7749,"longitude":-122.4194},"ts":"2026-08-20T07:08:09.123456789Z"}` |
| 66 | ++ batterLevelValue := `{"value":75.5,"ts":"2026-08-20T07:08:09.123456789Z"}` |
| 67 | + |
| 68 | + Expect(publishedTopics).To(HaveKey(vehicleNameTopic)) |
| 69 | + Expect(publishedTopics).To(HaveKey(invalidTopic)) |
0 commit comments