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

fix: throttle jt808 registration touches

Browse Source
This commit is contained in:
lingniu
2026-07-02 10:24:44 +08:00
parent a66f765e16
commit e7d024405a
6 changed files with 111 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
go/vehicle-gateway/cmd/gateway/main.go
View File

@@ -138,7 +138,10 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R
}
table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding")
logger.Info("identity mysql resolver enabled", "table", table)
return identity.NewMySQLResolver(db, table), func() { _ = db.Close() }, nil
locationTouchInterval := time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second
return identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{
LocationTouchInterval: locationTouchInterval,
}), func() { _ = db.Close() }, nil
}
func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) {

2
go/vehicle-gateway/internal/gateway/tcp_server_test.go
View File

@@ -37,7 +37,7 @@ func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) {
if len(sink.raw) != 1 || len(sink.unified) != 1 {
t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
}
if sink.raw[0].Phone != "013307795425" {
if sink.raw[0].Phone != "13307795425" {
t.Fatalf("phone = %q", sink.raw[0].Phone)
}
if sink.raw[0].Fields[envelope.FieldTotalMileageKM] != 10241.2 {

64
go/vehicle-gateway/internal/identity/resolver.go
View File

@@ -6,6 +6,8 @@ import (
"errors"
"fmt"
"strings"
"sync"
"time"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)
@@ -21,11 +23,22 @@ func (NoopResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (enve
}
type MySQLResolver struct {
db *sql.DB
table string
db *sql.DB
table string
locationTouchInterval time.Duration
touchMu sync.Mutex
locationTouches map[string]time.Time
}
type MySQLResolverOptions struct {
LocationTouchInterval time.Duration
}
func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver {
return NewMySQLResolverWithOptions(db, table, MySQLResolverOptions{})
}
func NewMySQLResolverWithOptions(db *sql.DB, table string, opts MySQLResolverOptions) *MySQLResolver {
if db == nil {
panic("identity db must not be nil")
}
@@ -33,7 +46,16 @@ func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver {
if table == "" || !safeIdentifier(table) {
table = "vehicle_identity_binding"
}
return &MySQLResolver{db: db, table: table}
interval := opts.LocationTouchInterval
if interval <= 0 {
interval = 10 * time.Minute
}
return &MySQLResolver{
db: db,
table: table,
locationTouchInterval: interval,
locationTouches: map[string]time.Time{},
}
}
func safeIdentifier(value string) bool {
@@ -50,6 +72,7 @@ func safeIdentifier(value string) bool {
}
func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
env.Phone = normalizePhone(env.Phone)
if strings.TrimSpace(env.VIN) != "" {
if err := r.trackJT808(ctx, env); err != nil {
return env, err
@@ -99,6 +122,9 @@ func (r *MySQLResolver) trackJT808(ctx context.Context, env envelope.FrameEnvelo
if env.Protocol != envelope.ProtocolJT808 || strings.TrimSpace(env.Phone) == "" || strings.TrimSpace(env.MessageID) == "" {
return nil
}
if env.MessageID != "0x0100" && env.MessageID != "0x0102" && !r.shouldTouchJT808Location(env) {
return nil
}
registration := mapValue(env.Parsed, "registration")
authentication := mapValue(env.Parsed, "authentication")
deviceID := firstNonEmpty(env.DeviceID, textValue(registration, "device_id"))
@@ -170,6 +196,25 @@ WHERE vin = ?`, env.Phone, env.Phone, deviceID, deviceID, plate, plate, vin)
return err
}
func (r *MySQLResolver) shouldTouchJT808Location(env envelope.FrameEnvelope) bool {
if env.MessageID != "0x0200" {
return false
}
phone := normalizePhone(env.Phone)
if phone == "" {
return false
}
now := time.Now()
r.touchMu.Lock()
defer r.touchMu.Unlock()
last, ok := r.locationTouches[phone]
if ok && now.Sub(last) < r.locationTouchInterval {
return false
}
r.locationTouches[phone] = now
return true
}
type CandidateKey struct {
Column string
Value string
@@ -189,16 +234,21 @@ func CandidateKeys(env envelope.FrameEnvelope) []CandidateKey {
}
out = append(out, CandidateKey{Column: column, Value: value})
}
add("phone", env.Phone)
if trimmed := strings.TrimLeft(strings.TrimSpace(env.Phone), "0"); trimmed != "" {
add("phone", trimmed)
}
add("phone", normalizePhone(env.Phone))
add("device_id", env.DeviceID)
add("plate", env.Plate)
add("vin", env.VehicleKeyHint)
return out
}
func normalizePhone(phone string) string {
trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0")
if trimmed == "" {
return strings.TrimSpace(phone)
}
return trimmed
}
func mapValue(values map[string]any, key string) map[string]any {
if values == nil {
return nil

46
go/vehicle-gateway/internal/identity/resolver_test.go
View File

@@ -21,7 +21,7 @@ func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) {
for _, key := range keys {
got = append(got, key.Column+"="+key.Value)
}
want := []string{"phone=013307795425", "phone=13307795425", "device_id=D1", "plate=豫A12345", "vin=D1"}
want := []string{"phone=13307795425", "device_id=D1", "plate=豫A12345", "vin=D1"}
if len(got) != len(want) {
t.Fatalf("candidate count = %d got=%#v", len(got), got)
}
@@ -32,7 +32,7 @@ func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) {
}
}
func TestCandidateKeysAddsTrimmedPhoneVariant(t *testing.T) {
func TestCandidateKeysNormalizesPhone(t *testing.T) {
keys := CandidateKeys(envelope.FrameEnvelope{
Phone: "013079963379",
})
@@ -40,7 +40,7 @@ func TestCandidateKeysAddsTrimmedPhoneVariant(t *testing.T) {
for _, key := range keys {
got = append(got, key.Column+"="+key.Value)
}
want := []string{"phone=013079963379", "phone=13079963379"}
want := []string{"phone=13079963379"}
if len(got) != len(want) {
t.Fatalf("candidate count = %d got=%#v", len(got), got)
}
@@ -55,7 +55,7 @@ func TestMySQLResolverFillsVINFromPhone(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("013307795425").
WithArgs("13307795425").
WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LNBVIN00000000001"))
resolver := NewMySQLResolver(db, "vehicle_identity_binding")
@@ -82,9 +82,6 @@ func TestMySQLResolverFillsVINFromPhone(t *testing.T) {
func TestMySQLResolverFallsBackToDeviceID(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("013307795425").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("13307795425").
WillReturnError(sql.ErrNoRows)
@@ -112,9 +109,6 @@ func TestMySQLResolverFallsBackToDeviceID(t *testing.T) {
func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("013079963379").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("13079963379").
WillReturnError(sql.ErrNoRows)
@@ -127,7 +121,7 @@ func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) {
mock.ExpectExec("INSERT INTO jt808_registration").
WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectExec("UPDATE IGNORE vehicle_identity_binding").
WithArgs("013079963379", "013079963379", "DEV0001", "DEV0001", "TEST123", "TEST123", "LKLG7C4E3NA774736").
WithArgs("13079963379", "13079963379", "DEV0001", "DEV0001", "TEST123", "TEST123", "LKLG7C4E3NA774736").
WillReturnResult(sqlmock.NewResult(0, 1))
resolver := NewMySQLResolver(db, "vehicle_identity_binding")
@@ -161,6 +155,36 @@ func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) {
}
}
func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("13307795425").
WillReturnError(sql.ErrNoRows)
mock.ExpectExec("INSERT INTO jt808_registration").
WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("13307795425").
WillReturnError(sql.ErrNoRows)
resolver := NewMySQLResolver(db, "vehicle_identity_binding")
for i := 0; i < 2; i++ {
_, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
MessageID: "0x0200",
Phone: "013307795425",
SourceEndpoint: "115.231.168.135:43625",
Parsed: map[string]any{},
})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func newMockDB(t *testing.T) (*sql.DB, sqlmock.Sqlmock) {
t.Helper()
db, mock, err := sqlmock.New()

17
go/vehicle-gateway/internal/protocol/jt808/parser.go
View File

@@ -112,11 +112,8 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
}
phoneBCD := raw[phoneStart:phoneEnd]
phone := bcdString(phoneBCD)
phoneTrimZero := strings.TrimLeft(phone, "0")
if versioned && phoneTrimZero != "" {
phone = phoneTrimZero
}
phoneRaw := bcdString(phoneBCD)
phone := normalizePhone(phoneRaw)
body := raw[headerLen : headerLen+int(bodySize)]
parsed := map[string]any{
"header": map[string]any{
@@ -127,7 +124,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
"body_size": bodySize,
"phone_bcd_hex": strings.ToUpper(hex.EncodeToString(phoneBCD)),
"phone": phone,
"phone_trim_zero": phoneTrimZero,
"phone_raw": phoneRaw,
"sequence": binary.BigEndian.Uint16(raw[serialStart : serialStart+2]),
"subpackage": subpackage,
"package_info": packageInfo,
@@ -230,6 +227,14 @@ func parseRegistration(body []byte, versioned bool) (map[string]any, error) {
return registration, nil
}
func normalizePhone(phone string) string {
trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0")
if trimmed == "" {
return strings.TrimSpace(phone)
}
return trimmed
}
func parseAuthentication(body []byte, versioned bool) map[string]any {
if versioned {
if out, ok := parseVersionedAuthentication(body); ok {

6
go/vehicle-gateway/internal/protocol/jt808/parser_test.go
View File

@@ -39,7 +39,7 @@ func TestExtractFramesUnescapesAndParsesLocationWithTotalMileage(t *testing.T) {
if env.MessageID != "0x0200" {
t.Fatalf("message id = %q", env.MessageID)
}
if env.Phone != "013307795425" {
if env.Phone != "13307795425" {
t.Fatalf("phone = %q", env.Phone)
}
if env.Sequence != 1 {
@@ -140,7 +140,7 @@ func TestParseFrameParsesCommonLocationAdditionalItems(t *testing.T) {
if err != nil {
t.Fatalf("ParseFrame() error = %v", err)
}
if env.Phone != "012345678901" {
if env.Phone != "12345678901" {
t.Fatalf("phone = %q", env.Phone)
}
assertFloatField(t, env, envelope.FieldTotalMileageKM, 1.1)
@@ -223,7 +223,7 @@ func TestParseFrameParsesRegistrationIdentity(t *testing.T) {
if env.MessageID != "0x0100" {
t.Fatalf("message id = %q", env.MessageID)
}
if env.Phone != "013079963379" {
if env.Phone != "13079963379" {
t.Fatalf("phone = %q", env.Phone)
}
if env.DeviceID != "DEV0001" {