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

feat(go): cache realtime locations in mysql

Browse Source
This commit is contained in:
lingniu
2026-07-02 16:31:58 +08:00
parent d0289b1b48
commit bc399d2819
2 changed files with 267 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,7 +26,10 @@ func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter {
}
func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error {
_, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL)
if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil {
return err
}
_, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL)
return err
}
@@ -48,7 +51,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope)
}
eventTime := nullableTime(env.EventTimeMS)
receivedAt := nullableTime(env.ReceivedAtMS)
_, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL,
if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL,
string(env.Protocol),
vehicleKey,
strings.TrimSpace(env.VIN),
@@ -63,10 +66,140 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope)
fieldsJSON,
receivedAt,
env.StableEventID(),
); err != nil {
return err
}
location, ok := realtimeLocationFromEnvelope(env, vehicleKey, fieldsJSON)
if !ok {
return nil
}
_, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL,
location.Protocol,
location.VehicleKey,
location.VIN,
location.Phone,
location.DeviceID,
location.Plate,
location.MessageID,
location.Sequence,
location.SourceEndpoint,
location.EventTime,
location.Latitude,
location.Longitude,
location.SpeedKMH,
location.TotalMileageKM,
location.SOCPercent,
location.AltitudeM,
location.DirectionDeg,
location.AlarmFlag,
location.StatusFlag,
location.FieldsJSON,
location.ReceivedAt,
location.EventID,
)
return err
}
type realtimeLocationRow struct {
Protocol string
VehicleKey string
VIN string
Phone string
DeviceID string
Plate string
MessageID string
Sequence uint16
SourceEndpoint string
EventTime any
Latitude float64
Longitude float64
SpeedKMH any
TotalMileageKM any
SOCPercent any
AltitudeM any
DirectionDeg any
AlarmFlag any
StatusFlag any
FieldsJSON string
ReceivedAt any
EventID string
}
func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, fieldsJSON string) (realtimeLocationRow, bool) {
latitude, okLat := numberField(env.Fields, envelope.FieldLatitude)
longitude, okLon := numberField(env.Fields, envelope.FieldLongitude)
if !okLat || !okLon {
return realtimeLocationRow{}, false
}
return realtimeLocationRow{
Protocol: string(env.Protocol),
VehicleKey: vehicleKey,
VIN: strings.TrimSpace(env.VIN),
Phone: strings.TrimSpace(env.Phone),
DeviceID: strings.TrimSpace(env.DeviceID),
Plate: strings.TrimSpace(env.Plate),
MessageID: strings.TrimSpace(env.MessageID),
Sequence: env.Sequence,
SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
EventTime: nullableTime(env.EventTimeMS),
Latitude: latitude,
Longitude: longitude,
SpeedKMH: nullableNumberField(env.Fields, envelope.FieldSpeedKMH),
TotalMileageKM: nullableNumberField(env.Fields, envelope.FieldTotalMileageKM),
SOCPercent: nullableNumberField(env.Fields, envelope.FieldSOCPercent),
AltitudeM: nullableNumberField(env.Fields, "altitude_m"),
DirectionDeg: nullableNumberField(env.Fields, "direction_deg"),
AlarmFlag: nullableNumberField(env.Fields, "alarm_flag"),
StatusFlag: nullableNumberField(env.Fields, "status_flag"),
FieldsJSON: fieldsJSON,
ReceivedAt: nullableTime(env.ReceivedAtMS),
EventID: env.StableEventID(),
}, true
}
func nullableNumberField(fields map[string]any, key string) any {
value, ok := numberField(fields, key)
if !ok {
return nil
}
return value
}
func numberField(fields map[string]any, key string) (float64, bool) {
value, ok := fields[key]
if !ok {
return 0, false
}
switch typed := value.(type) {
case float64:
return typed, true
case float32:
return float64(typed), true
case int:
return float64(typed), true
case int8:
return float64(typed), true
case int16:
return float64(typed), true
case int32:
return float64(typed), true
case int64:
return float64(typed), true
case uint:
return float64(typed), true
case uint8:
return float64(typed), true
case uint16:
return float64(typed), true
case uint32:
return float64(typed), true
case uint64:
return float64(typed), true
default:
return 0, false
}
}
func marshalObject(value map[string]any) (string, error) {
if value == nil {
value = map[string]any{}
@@ -139,3 +272,67 @@ ON DUPLICATE KEY UPDATE
event_id = VALUES(event_id),
updated_at = CURRENT_TIMESTAMP
`
const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
protocol VARCHAR(32) NOT NULL,
vehicle_key VARCHAR(96) NOT NULL,
vin VARCHAR(32) NOT NULL DEFAULT '',
phone VARCHAR(32) NOT NULL DEFAULT '',
device_id VARCHAR(96) NOT NULL DEFAULT '',
plate VARCHAR(32) NOT NULL DEFAULT '',
message_id VARCHAR(32) NOT NULL DEFAULT '',
sequence_id INT NOT NULL DEFAULT 0,
source_endpoint VARCHAR(128) NOT NULL DEFAULT '',
event_time DATETIME(3) NULL,
latitude DECIMAL(12,6) NOT NULL,
longitude DECIMAL(12,6) NOT NULL,
speed_kmh DECIMAL(10,3) NULL,
total_mileage_km DECIMAL(18,3) NULL,
soc_percent DECIMAL(6,2) NULL,
altitude_m DECIMAL(10,3) NULL,
direction_deg DECIMAL(10,3) NULL,
alarm_flag BIGINT NULL,
status_flag BIGINT NULL,
fields_json JSON NOT NULL,
received_at DATETIME(3) NULL,
event_id VARCHAR(64) NOT NULL DEFAULT '',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_realtime_location_vehicle (protocol, vehicle_key),
KEY idx_vehicle_key (vehicle_key),
KEY idx_vin (vin),
KEY idx_protocol_updated (protocol, updated_at),
KEY idx_location (longitude, latitude)
)`
const upsertRealtimeLocationSQL = `
INSERT INTO vehicle_realtime_location
(protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id,
source_endpoint, event_time, latitude, longitude, speed_kmh, total_mileage_km,
soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, fields_json,
received_at, event_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?)
ON DUPLICATE KEY UPDATE
vin = IF(VALUES(vin) <> '', VALUES(vin), vin),
phone = IF(VALUES(phone) <> '', VALUES(phone), phone),
device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id),
plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
message_id = VALUES(message_id),
sequence_id = VALUES(sequence_id),
source_endpoint = VALUES(source_endpoint),
event_time = VALUES(event_time),
latitude = VALUES(latitude),
longitude = VALUES(longitude),
speed_kmh = VALUES(speed_kmh),
total_mileage_km = VALUES(total_mileage_km),
soc_percent = VALUES(soc_percent),
altitude_m = VALUES(altitude_m),
direction_deg = VALUES(direction_deg),
alarm_flag = VALUES(alarm_flag),
status_flag = VALUES(status_flag),
fields_json = VALUES(fields_json),
received_at = VALUES(received_at),
event_id = VALUES(event_id),
updated_at = CURRENT_TIMESTAMP
`

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

@@ -36,13 +36,16 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 2 {
t.Fatalf("exec calls = %d, want 2", len(exec.calls))
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want 3", len(exec.calls))
}
if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") {
t.Fatalf("schema query = %s", exec.calls[0].query)
}
upsert := exec.calls[1]
if !strings.Contains(exec.calls[1].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") {
t.Fatalf("location schema query = %s", exec.calls[1].query)
}
upsert := exec.calls[2]
if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("upsert query = %s", upsert.query)
}
@@ -61,6 +64,68 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) {
}
}
func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)
if err := writer.Update(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
MessageID: "0x0200",
Sequence: 9,
VIN: "VIN001",
Phone: "13307795425",
DeviceID: "device-1",
Plate: "沪A12345",
SourceEndpoint: "1.2.3.4:808",
EventTimeMS: 1782918600000,
ReceivedAtMS: 1782918601000,
EventID: "event-1",
Parsed: map[string]any{"location": map[string]any{"latitude": 30.590151, "longitude": 121.069881}},
Fields: map[string]any{
envelope.FieldLatitude: 30.590151,
envelope.FieldLongitude: 121.069881,
envelope.FieldSpeedKMH: 23.0,
envelope.FieldTotalMileageKM: 10241.2,
envelope.FieldSOCPercent: 88,
"altitude_m": uint16(5),
"direction_deg": uint16(79),
"alarm_flag": uint32(0),
"status_flag": uint32(786435),
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 2 {
t.Fatalf("exec calls = %d, want 2", len(exec.calls))
}
locationUpsert := exec.calls[1]
if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") {
t.Fatalf("location upsert query = %s", locationUpsert.query)
}
if got, want := locationUpsert.args[0], "JT808"; got != want {
t.Fatalf("protocol arg = %#v, want %q", got, want)
}
if got, want := locationUpsert.args[1], "VIN001"; got != want {
t.Fatalf("vehicle key arg = %#v, want %q", got, want)
}
if got, want := locationUpsert.args[10], 30.590151; got != want {
t.Fatalf("latitude arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[11], 121.069881; got != want {
t.Fatalf("longitude arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[12], 23.0; got != want {
t.Fatalf("speed arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[13], 10241.2; got != want {
t.Fatalf("mileage arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[14], 88.0; got != want {
t.Fatalf("soc arg = %#v, want %v", got, want)
}
}
func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) {
exec := &recordingSnapshotExec{}
writer := NewSnapshotWriter(exec)