fix: stabilize yutong mqtt ingestion

This commit is contained in:
lingniu
2026-07-02 01:01:33 +08:00
parent 6d1d0aa85e
commit 3ee1d6f52b
4 changed files with 27 additions and 4 deletions

View File

@@ -108,6 +108,8 @@ func (c *MQTTClient) buildOptions(ctx context.Context) (*mqtt.ClientOptions, err
SetAutoReconnect(true). SetAutoReconnect(true).
SetConnectRetry(true). SetConnectRetry(true).
SetConnectRetryInterval(5 * time.Second). SetConnectRetryInterval(5 * time.Second).
SetResumeSubs(true).
SetOrderMatters(false).
SetKeepAlive(keepAlive). SetKeepAlive(keepAlive).
SetConnectTimeout(connectTimeout) SetConnectTimeout(connectTimeout)

View File

@@ -111,6 +111,12 @@ func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) {
if !opts.CleanSession { if !opts.CleanSession {
t.Fatal("CleanSession should be true") 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 { if got := opts.KeepAlive; got != 20 {
t.Fatalf("KeepAlive = %d, want 20", got) t.Fatalf("KeepAlive = %d, want 20", got)
} }

View File

@@ -133,7 +133,7 @@ func rawValues(env envelope.FrameEnvelope) []any {
messageIDInt(env.MessageID), messageIDInt(env.MessageID),
eventTime, eventTime,
received, received,
rawSize(env.RawHex), rawSizeBytes(env),
env.RawHex, env.RawHex,
env.RawText, env.RawText,
jsonString(env.Parsed), jsonString(env.Parsed),
@@ -200,13 +200,16 @@ func messageIDInt(value string) int64 {
return parsed return parsed
} }
func rawSize(rawHex string) int { func rawSizeBytes(env envelope.FrameEnvelope) int {
rawHex = strings.TrimSpace(rawHex) rawHex := strings.TrimSpace(env.RawHex)
if len(rawHex)%2 != 0 { if len(rawHex)%2 != 0 {
return len(rawHex) / 2 return len(rawHex) / 2
} }
if rawHex != "" {
return len(rawHex) / 2 return len(rawHex) / 2
} }
return len([]byte(env.RawText))
}
func jsonString(value any) string { func jsonString(value any) string {
data, err := json.Marshal(value) data, err := json.Marshal(value)

View File

@@ -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 { func sampleEnvelope() envelope.FrameEnvelope {
return envelope.FrameEnvelope{ return envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808, Protocol: envelope.ProtocolJT808,