diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 40289aba..99f7bed7 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -108,6 +108,8 @@ func (c *MQTTClient) buildOptions(ctx context.Context) (*mqtt.ClientOptions, err SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(5 * time.Second). + SetResumeSubs(true). + SetOrderMatters(false). SetKeepAlive(keepAlive). SetConnectTimeout(connectTimeout) diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index 1f21fd26..a5cbd696 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -111,6 +111,12 @@ func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) { if !opts.CleanSession { t.Fatal("CleanSession should be true") } + if opts.Order { + t.Fatal("OrderMatters should be false so MQTT network handling is not blocked by Kafka/DB work") + } + if !opts.ResumeSubs { + t.Fatal("ResumeSubs should be true to avoid subscribe failures during reconnect churn") + } if got := opts.KeepAlive; got != 20 { t.Fatalf("KeepAlive = %d, want 20", got) } diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index c58f6b25..e54b4553 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -133,7 +133,7 @@ func rawValues(env envelope.FrameEnvelope) []any { messageIDInt(env.MessageID), eventTime, received, - rawSize(env.RawHex), + rawSizeBytes(env), env.RawHex, env.RawText, jsonString(env.Parsed), @@ -200,12 +200,15 @@ func messageIDInt(value string) int64 { return parsed } -func rawSize(rawHex string) int { - rawHex = strings.TrimSpace(rawHex) +func rawSizeBytes(env envelope.FrameEnvelope) int { + rawHex := strings.TrimSpace(env.RawHex) if len(rawHex)%2 != 0 { return len(rawHex) / 2 } - return len(rawHex) / 2 + if rawHex != "" { + return len(rawHex) / 2 + } + return len([]byte(env.RawText)) } func jsonString(value any) string { diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index ddc27c61..255d94e5 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -75,6 +75,18 @@ func TestWriterSkipsSparseDerivedRows(t *testing.T) { } } +func TestRawSizeBytesUsesRawTextWhenHexIsEmpty(t *testing.T) { + env := envelope.FrameEnvelope{RawText: `{"code":"0F80","data":{"speed":12}}`} + if got, want := rawSizeBytes(env), len([]byte(env.RawText)); got != want { + t.Fatalf("raw text size = %d, want %d", got, want) + } + + env.RawHex = "7E0200" + if got, want := rawSizeBytes(env), 3; got != want { + t.Fatalf("raw hex size = %d, want %d", got, want) + } +} + func sampleEnvelope() envelope.FrameEnvelope { return envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808,