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

feat: expose realtime snapshot protocols

Browse Source
This commit is contained in:
lingniu
2026-07-02 01:23:03 +08:00
parent eb124d929b
commit a7b24f4c0b
3 changed files with 28 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,16 +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"`
EventID string `json:"event_id,omitempty"` Protocols []envelope.Protocol `json:"protocols,omitempty"`
EventTimeMS int64 `json:"event_time_ms"` EventID string `json:"event_id,omitempty"`
ReceivedAtMS int64 `json:"received_at_ms"` EventTimeMS int64 `json:"event_time_ms"`
SourceEndpoint string `json:"source_endpoint,omitempty"` ReceivedAtMS int64 `json:"received_at_ms"`
Fields map[string]any `json:"fields,omitempty"` SourceEndpoint string `json:"source_endpoint,omitempty"`
FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"` Fields map[string]any `json:"fields,omitempty"`
UpdatedAtMS int64 `json:"updated_at_ms"` FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"`
UpdatedAtMS int64 `json:"updated_at_ms"`
} }
type OnlineStatus struct { type OnlineStatus struct {

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

@@ -53,6 +53,11 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
return err return err
} }
protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol)
if err != nil {
return err
}
merged, err := r.GetMerged(ctx, vehicleKey) merged, err := r.GetMerged(ctx, vehicleKey)
if err != nil && !errors.Is(err, redis.Nil) { if err != nil && !errors.Is(err, redis.Nil) {
return err return err
@@ -69,15 +74,12 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
merged.ReceivedAtMS = env.ReceivedAtMS merged.ReceivedAtMS = env.ReceivedAtMS
merged.SourceEndpoint = env.SourceEndpoint merged.SourceEndpoint = env.SourceEndpoint
} }
merged.Protocols = protocols
merged.UpdatedAtMS = nowMS merged.UpdatedAtMS = nowMS
if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil { if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil {
return err return err
} }
protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol)
if err != nil {
return err
}
online := OnlineStatus{ online := OnlineStatus{
VehicleKey: vehicleKey, VehicleKey: vehicleKey,
VIN: vin, VIN: vin,

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

@@ -54,6 +54,9 @@ func TestRepositoryUpdatesMergedAndProtocolSnapshots(t *testing.T) {
if merged.Fields[envelope.FieldLatitude] != 30.5 { if merged.Fields[envelope.FieldLatitude] != 30.5 {
t.Fatalf("new latitude missing: %#v", merged.Fields) t.Fatalf("new latitude missing: %#v", merged.Fields)
} }
if got := protocolNames(merged.Protocols); strings.Join(got, ",") != "GB32960,JT808" {
t.Fatalf("merged protocols = %#v", got)
}
protocol, err := repo.GetProtocol(ctx, "VIN001", envelope.ProtocolJT808) protocol, err := repo.GetProtocol(ctx, "VIN001", envelope.ProtocolJT808)
if err != nil { if err != nil {
t.Fatalf("GetProtocol() error = %v", err) t.Fatalf("GetProtocol() error = %v", err)
@@ -237,3 +240,11 @@ func newTestRepository(t *testing.T) (*Repository, func()) {
func stringsContains(value string, pattern string) bool { func stringsContains(value string, pattern string) bool {
return strings.Contains(value, pattern) return strings.Contains(value, pattern)
} }
func protocolNames(protocols []envelope.Protocol) []string {
names := make([]string, 0, len(protocols))
for _, protocol := range protocols {
names = append(names, string(protocol))
}
return names
}