shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(go): filter realtime mysql snapshots by business events

Browse Source
This commit is contained in:
lingniu
2026-07-03 10:42:32 +08:00
parent 02075b5ea0
commit 9c212815ef
2 changed files with 107 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
go/vehicle-gateway/internal/realtime/snapshot_writer.go
View File

@@ -119,7 +119,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope)
if vin == "" {
return nil
}
if !hasRealtimePayload(env) {
if !isRealtimeSnapshotEvent(env) {
return nil
}
plate, err := w.plateForEnvelope(ctx, env)
@@ -365,16 +365,35 @@ func nullableTime(ms int64) any {
return time.UnixMilli(ms)
}
func hasRealtimePayload(env envelope.FrameEnvelope) bool {
if len(env.Fields) > 0 {
return true
func isRealtimeSnapshotEvent(env envelope.FrameEnvelope) bool {
switch env.Protocol {
case envelope.ProtocolGB32960:
return env.MessageID == "0x02" && hasGB32960RealtimeDataUnits(env)
case envelope.ProtocolJT808:
return env.MessageID == "0x0200" && hasLocationFields(env)
case envelope.ProtocolYutongMQTT:
return len(env.Fields) > 0
default:
return false
}
}
func hasGB32960RealtimeDataUnits(env envelope.FrameEnvelope) bool {
if units, ok := env.Parsed["data_units"].([]any); ok && len(units) > 0 {
return true
}
if units, ok := env.Parsed["data_units"].([]map[string]any); ok && len(units) > 0 {
return true
}
return false
}
func hasLocationFields(env envelope.FrameEnvelope) bool {
_, okLat := numberField(env.Fields, envelope.FieldLatitude)
_, okLon := numberField(env.Fields, envelope.FieldLongitude)
return okLat && okLon
}
const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot (
protocol VARCHAR(32) NOT NULL,
vin VARCHAR(32) NOT NULL DEFAULT '',

87
go/vehicle-gateway/internal/realtime/snapshot_writer_test.go
View File

@@ -234,7 +234,7 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) {
VIN: "VIN001",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{"data_units": []any{}},
Parsed: sampleGB32960RealtimeParsed(),
Fields: map[string]any{
envelope.FieldLatitude: 30.590151,
envelope.FieldLongitude: 121.069881,
@@ -289,6 +289,7 @@ func TestSnapshotWriterCachesBindingPlateByVIN(t *testing.T) {
VIN: "VIN001",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: sampleGB32960RealtimeParsed(),
Fields: map[string]any{
envelope.FieldLatitude: 30.590151,
envelope.FieldLongitude: 121.069881,
@@ -355,7 +356,7 @@ func TestSnapshotWriterIgnoresMissingBindingPlate(t *testing.T) {
VIN: "VIN001",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{"data_units": []any{}},
Parsed: sampleGB32960RealtimeParsed(),
Fields: map[string]any{envelope.FieldSOCPercent: 90},
}); err != nil {
t.Fatalf("Update() error = %v", err)
@@ -375,7 +376,7 @@ func TestSnapshotWriterReturnsUnexpectedPlateLookupError(t *testing.T) {
VIN: "VIN001",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{"data_units": []any{}},
Parsed: sampleGB32960RealtimeParsed(),
Fields: map[string]any{envelope.FieldSOCPercent: 90},
})
if err == nil || !strings.Contains(err.Error(), "db down") {
@@ -445,6 +446,78 @@ func TestSnapshotWriterSkipsGB32960HeartbeatWithoutRealtimePayload(t *testing.T)
}
}
func TestSnapshotWriterSkipsGB32960NonRealtimeEvenWithConnectionMetadata(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)
if err := writer.Update(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolGB32960,
MessageID: "0x07",
VIN: "ABCDE600000000009",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{
"header": map[string]any{"command": "0x07", "vin": "ABCDE600000000009"},
"platform_name": "YueJin",
},
Fields: map[string]any{
"platform_account": "YueJin",
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 0 {
t.Fatalf("exec calls = %d, want 0", len(exec.calls))
}
}
func TestSnapshotWriterSkipsJT808NonLocationEvenWithIdentityFields(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)
if err := writer.Update(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
MessageID: "0x0100",
VIN: "VIN001",
Plate: "粤A12345",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{
"registration": map[string]any{"plate": "粤A12345"},
},
Fields: map[string]any{
"plate": "粤A12345",
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 0 {
t.Fatalf("exec calls = %d, want 0", len(exec.calls))
}
}
func TestSnapshotWriterSkipsMQTTWithoutActualDataFields(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)
if err := writer.Update(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolYutongMQTT,
MessageID: "MQTT",
VIN: "VIN001",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
Parsed: map[string]any{
"topic": "/vehicle/VIN001/state",
"data": map[string]any{},
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 0 {
t.Fatalf("exec calls = %d, want 0", len(exec.calls))
}
}
func TestSnapshotWriterSkipsEmptyDataUnitsWithoutCoreFields(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)
@@ -468,6 +541,14 @@ type recordingSnapshotExec struct {
calls []snapshotExecCall
}
func sampleGB32960RealtimeParsed() map[string]any {
return map[string]any{
"data_units": []any{
map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 90}},
},
}
}
type snapshotExecCall struct {
query string
args []any