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

refactor(go): remove source endpoint from realtime snapshots

Browse Source
This commit is contained in:
lingniu
2026-07-02 21:16:39 +08:00
parent 04ca0d15d8
commit ef2be36fce
3 changed files with 30 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
go/vehicle-gateway/internal/realtime/model.go
View File

@@ -7,18 +7,17 @@ import (
) )
type Snapshot struct { type Snapshot struct {
VehicleKey string `json:"vehicle_key"` VehicleKey string `json:"vehicle_key"`
VIN string `json:"vin"` VIN string `json:"vin"`
Protocol envelope.Protocol `json:"protocol,omitempty"` Protocol envelope.Protocol `json:"protocol,omitempty"`
Protocols []envelope.Protocol `json:"protocols,omitempty"` Protocols []envelope.Protocol `json:"protocols,omitempty"`
EventID string `json:"event_id,omitempty"` EventID string `json:"event_id,omitempty"`
EventTimeMS int64 `json:"event_time_ms"` EventTimeMS int64 `json:"event_time_ms"`
ReceivedAtMS int64 `json:"received_at_ms"` ReceivedAtMS int64 `json:"received_at_ms"`
SourceEndpoint string `json:"source_endpoint,omitempty"` Fields map[string]any `json:"fields,omitempty"`
Fields map[string]any `json:"fields,omitempty"` FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"`
FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"` Parsed map[string]any `json:"parsed,omitempty"`
Parsed map[string]any `json:"parsed,omitempty"` UpdatedAtMS int64 `json:"updated_at_ms"`
UpdatedAtMS int64 `json:"updated_at_ms"`
} }
func (s Snapshot) Lightweight() Snapshot { func (s Snapshot) Lightweight() Snapshot {

23
go/vehicle-gateway/internal/realtime/repository.go
View File

@@ -39,17 +39,16 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
eventMS = env.ReceivedAtMS eventMS = env.ReceivedAtMS
} }
protocolSnapshot := Snapshot{ protocolSnapshot := Snapshot{
VehicleKey: vehicleKey, VehicleKey: vehicleKey,
VIN: vin, VIN: vin,
Protocol: env.Protocol, Protocol: env.Protocol,
EventID: env.StableEventID(), EventID: env.StableEventID(),
EventTimeMS: eventMS, EventTimeMS: eventMS,
ReceivedAtMS: env.ReceivedAtMS, ReceivedAtMS: env.ReceivedAtMS,
SourceEndpoint: env.SourceEndpoint, Fields: cloneFields(env.Fields),
Fields: cloneFields(env.Fields), FieldTimesMS: fieldTimes(env.Fields, eventMS),
FieldTimesMS: fieldTimes(env.Fields, eventMS), Parsed: cloneMap(env.Parsed),
Parsed: cloneMap(env.Parsed), UpdatedAtMS: nowMS,
UpdatedAtMS: nowMS,
} }
existingProtocol, err := r.GetProtocol(ctx, vehicleKey, env.Protocol) existingProtocol, err := r.GetProtocol(ctx, vehicleKey, env.Protocol)
if err != nil && !errors.Is(err, redis.Nil) { if err != nil && !errors.Is(err, redis.Nil) {
@@ -69,7 +68,6 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
protocolSnapshot.EventTimeMS = eventMS protocolSnapshot.EventTimeMS = eventMS
protocolSnapshot.EventID = env.StableEventID() protocolSnapshot.EventID = env.StableEventID()
protocolSnapshot.ReceivedAtMS = env.ReceivedAtMS protocolSnapshot.ReceivedAtMS = env.ReceivedAtMS
protocolSnapshot.SourceEndpoint = env.SourceEndpoint
} }
protocolSnapshot.UpdatedAtMS = nowMS protocolSnapshot.UpdatedAtMS = nowMS
} }
@@ -105,7 +103,6 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
merged.EventTimeMS = eventMS merged.EventTimeMS = eventMS
merged.EventID = env.StableEventID() merged.EventID = env.StableEventID()
merged.ReceivedAtMS = env.ReceivedAtMS merged.ReceivedAtMS = env.ReceivedAtMS
merged.SourceEndpoint = env.SourceEndpoint
} }
merged.Protocols = protocols merged.Protocols = protocols
merged.UpdatedAtMS = nowMS merged.UpdatedAtMS = nowMS

14
go/vehicle-gateway/internal/realtime/repository_test.go
View File

@@ -380,11 +380,12 @@ func TestHandlerReturnsMergedSnapshot(t *testing.T) {
repo, closeFn := newTestRepository(t) repo, closeFn := newTestRepository(t)
defer closeFn() defer closeFn()
if err := repo.Update(context.Background(), envelope.FrameEnvelope{ if err := repo.Update(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808, Protocol: envelope.ProtocolJT808,
VIN: "VIN001", VIN: "VIN001",
EventTimeMS: 1000, SourceEndpoint: "115.231.168.135:43625",
ReceivedAtMS: 1100, EventTimeMS: 1000,
Fields: map[string]any{envelope.FieldSpeedKMH: 12.3}, ReceivedAtMS: 1100,
Fields: map[string]any{envelope.FieldSpeedKMH: 12.3},
}); err != nil { }); err != nil {
t.Fatalf("Update() error = %v", err) t.Fatalf("Update() error = %v", err)
} }
@@ -398,6 +399,9 @@ func TestHandlerReturnsMergedSnapshot(t *testing.T) {
if !stringsContains(rec.Body.String(), `"vin":"VIN001"`) { if !stringsContains(rec.Body.String(), `"vin":"VIN001"`) {
t.Fatalf("unexpected body: %s", rec.Body.String()) t.Fatalf("unexpected body: %s", rec.Body.String())
} }
if stringsContains(rec.Body.String(), "source_endpoint") {
t.Fatalf("realtime snapshot should not expose source_endpoint: %s", rec.Body.String())
}
} }
func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) {