From 114e707a5b81edd83f256ff3dccca69d3093062c Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 16:53:55 +0800 Subject: [PATCH] feat: add mysql vehicle identity store --- deploy/local/launchctl/README.md | 8 + .../com.lingniu.jt808.plist.template | 8 + deploy/portainer/docker-compose.yml | 5 + .../vehicle-ingest-tdengine-verification.md | 10 +- .../src/main/resources/application.yml | 6 + .../jt808app/Jt808IngestAppDefaultsTest.java | 4 + modules/core/vehicle-identity/pom.xml | 5 + .../identity/FileVehicleIdentityService.java | 3 + .../InMemoryVehicleIdentityService.java | 3 + .../identity/MySqlVehicleIdentityService.java | 145 +++++++++++++ .../identity/VehicleIdentityBinding.java | 4 + .../VehicleIdentityAutoConfiguration.java | 82 ++++++++ .../config/VehicleIdentityProperties.java | 22 ++ .../MySqlVehicleIdentityServiceTest.java | 199 ++++++++++++++++++ .../VehicleIdentityAutoConfigurationTest.java | 60 ++++++ .../jt808/inbound/Jt808ChannelHandler.java | 4 +- .../inbound/Jt808ChannelHandlerTest.java | 41 ++++ 17 files changed, 607 insertions(+), 2 deletions(-) create mode 100644 modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java create mode 100644 modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java diff --git a/deploy/local/launchctl/README.md b/deploy/local/launchctl/README.md index 74142ecd..b15caf73 100644 --- a/deploy/local/launchctl/README.md +++ b/deploy/local/launchctl/README.md @@ -20,6 +20,9 @@ export EVENT_STORE_PATH="$PROJECT_ROOT/data/event-store-live" export TDENGINE_JDBC_URL='jdbc:TAOS-WS://:6041/vehicle_ts' export TDENGINE_USERNAME='root' export TDENGINE_PASSWORD='' +export VEHICLE_IDENTITY_MYSQL_JDBC_URL='jdbc:mysql://:3306/vehicle_ingest' +export VEHICLE_IDENTITY_MYSQL_USERNAME='' +export VEHICLE_IDENTITY_MYSQL_PASSWORD='' mkdir -p "$HOME/Library/LaunchAgents" "$PROJECT_ROOT/data" mkdir -p /tmp/lingniu-gb32960-live /tmp/lingniu-jt808-live /tmp/lingniu-history-live @@ -33,6 +36,9 @@ for service in gb32960 jt808 vehicle-history; do -e "s#__TDENGINE_JDBC_URL__#$TDENGINE_JDBC_URL#g" \ -e "s#__TDENGINE_USERNAME__#$TDENGINE_USERNAME#g" \ -e "s#__TDENGINE_PASSWORD__#$TDENGINE_PASSWORD#g" \ + -e "s#__VEHICLE_IDENTITY_MYSQL_JDBC_URL__#$VEHICLE_IDENTITY_MYSQL_JDBC_URL#g" \ + -e "s#__VEHICLE_IDENTITY_MYSQL_USERNAME__#$VEHICLE_IDENTITY_MYSQL_USERNAME#g" \ + -e "s#__VEHICLE_IDENTITY_MYSQL_PASSWORD__#$VEHICLE_IDENTITY_MYSQL_PASSWORD#g" \ "deploy/local/launchctl/com.lingniu.${service}.plist.template" \ > "$HOME/Library/LaunchAgents/com.lingniu.${service}.plist" done @@ -40,6 +46,8 @@ done 如果生产配置由 Nacos 下发，可以把模板里的 `NACOS_CONFIG_ENABLED` 改成 `true`，并补充 `NACOS_SERVER_ADDR`、`NACOS_NAMESPACE`、`NACOS_GROUP`、`NACOS_USERNAME`、`NACOS_PASSWORD`。端口、Kafka topic、archive root、TDengine 连接建议仍保留为启动环境变量，便于 Portainer、launchctl 和临时压测保持一致。 +JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 表并按 phone/deviceId/plate 维护 VIN 绑定。 + ## 启停 ```bash diff --git a/deploy/local/launchctl/com.lingniu.jt808.plist.template b/deploy/local/launchctl/com.lingniu.jt808.plist.template index ce99c07f..731bc532 100644 --- a/deploy/local/launchctl/com.lingniu.jt808.plist.template +++ b/deploy/local/launchctl/com.lingniu.jt808.plist.template @@ -34,6 +34,14 @@ __SHARED_ARCHIVE_PATH__ VEHICLE_IDENTITY_STORE memory + VEHICLE_IDENTITY_MYSQL_TABLE + vehicle_identity_binding + VEHICLE_IDENTITY_MYSQL_JDBC_URL + __VEHICLE_IDENTITY_MYSQL_JDBC_URL__ + VEHICLE_IDENTITY_MYSQL_USERNAME + __VEHICLE_IDENTITY_MYSQL_USERNAME__ + VEHICLE_IDENTITY_MYSQL_PASSWORD + __VEHICLE_IDENTITY_MYSQL_PASSWORD__ NACOS_CONFIG_ENABLED false MANAGEMENT_HEALTH_REDIS_ENABLED diff --git a/deploy/portainer/docker-compose.yml b/deploy/portainer/docker-compose.yml index 8fe81589..34f14275 100644 --- a/deploy/portainer/docker-compose.yml +++ b/deploy/portainer/docker-compose.yml @@ -61,7 +61,12 @@ services: KAFKA_CONSUMER_ENABLED: "false" KAFKA_NODE_ID: ${KAFKA_NODE_ID_JT808:-jt808-ingest-portainer} SINK_ARCHIVE_PATH: /archive/ + VEHICLE_IDENTITY_STORE: ${VEHICLE_IDENTITY_STORE:-file} VEHICLE_IDENTITY_FILE: /data/vehicle-identity.jsonl + VEHICLE_IDENTITY_MYSQL_TABLE: ${VEHICLE_IDENTITY_MYSQL_TABLE:-vehicle_identity_binding} + VEHICLE_IDENTITY_MYSQL_JDBC_URL: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:-} + VEHICLE_IDENTITY_MYSQL_USERNAME: ${VEHICLE_IDENTITY_MYSQL_USERNAME:-} + VEHICLE_IDENTITY_MYSQL_PASSWORD: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:-} ports: - "${JT808_HTTP_PORT:-20400}:20400" - "${JT808_TCP_PORT:-808}:808" diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 53c80d58..f17cde8e 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -109,6 +109,14 @@ deploy/local/launchctl/ - 三个服务必须共用同一个 `SINK_ARCHIVE_PATH`，否则 TDengine 中的 `raw_uri` 能入库，但 history API 无法按 `archive://...` 读取原始帧。 - 替换 jar 前先 `launchctl bootout`，再复制 jar 和 `launchctl bootstrap`，避免 KeepAlive 在 jar 拷贝中途重启。 +JT808 注册身份绑定： + +- 默认可以使用 `VEHICLE_IDENTITY_STORE=file` 或 `memory`。 +- 生产需要把 0x0100 注册信息维护到 MySQL 时，设置 `VEHICLE_IDENTITY_STORE=mysql`。 +- 需要提供 `VEHICLE_IDENTITY_MYSQL_JDBC_URL`、`VEHICLE_IDENTITY_MYSQL_USERNAME`、`VEHICLE_IDENTITY_MYSQL_PASSWORD`。 +- 服务启动时自动创建 `vehicle_identity_binding` 表，按 `protocol + identifier_type + identifier_value` 维护 VIN 绑定。 +- 绑定类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定 VIN。 + 健康检查： ```bash @@ -271,4 +279,4 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots/fields?vin= - 当前本机使用 TDengine WebSocket JDBC，例如 `jdbc:TAOS-WS://:6041/vehicle_ts`。 - `KAFKA_CONSUMER_AUTO_OFFSET_RESET=latest` 适合生产接入新流量；如果要回放历史 Kafka 数据，需要切换 consumer group 或重置 offset。 - `vehicle-history-app` raw 和 event 使用不同 consumer binding；raw consumer 必须开启，否则 `raw_frames` 不会持续增长。 -- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:`，后续维护注册/VIN 映射后再反写 VIN。 +- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:`；启用 MySQL identity store 后，注册维护的 phone/deviceId/plate 绑定会用于后续 raw/event 的 VIN 解析。 diff --git a/modules/apps/jt808-ingest-app/src/main/resources/application.yml b/modules/apps/jt808-ingest-app/src/main/resources/application.yml index a91da2f6..cb83c7aa 100644 --- a/modules/apps/jt808-ingest-app/src/main/resources/application.yml +++ b/modules/apps/jt808-ingest-app/src/main/resources/application.yml @@ -51,6 +51,12 @@ lingniu: store: ${VEHICLE_IDENTITY_STORE:file} file: path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl} + mysql: + table: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding} + jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:} + username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:} + password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:} + driver-class-name: ${VEHICLE_IDENTITY_MYSQL_DRIVER_CLASS_NAME:com.mysql.cj.jdbc.Driver} sink: mq: enabled: ${KAFKA_ENABLED:true} diff --git a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java index 575b9a59..281c54b2 100644 --- a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java +++ b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java @@ -28,6 +28,10 @@ class Jt808IngestAppDefaultsTest { .containsEntry("lingniu.ingest.sink.mq.topics.realtime", "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}") .containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}") .containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1}") + .containsEntry("lingniu.ingest.identity.store", "${VEHICLE_IDENTITY_STORE:file}") + .containsEntry("lingniu.ingest.identity.mysql.table", "${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}") + .containsEntry("lingniu.ingest.identity.mysql.jdbc-url", "${VEHICLE_IDENTITY_MYSQL_JDBC_URL:}") + .containsEntry("lingniu.ingest.identity.mysql.username", "${VEHICLE_IDENTITY_MYSQL_USERNAME:}") .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}") .containsEntry("lingniu.ingest.sink.archive.path", "${SINK_ARCHIVE_PATH:./archive/}") .containsEntry("lingniu.ingest.event-file-store.enabled", false) diff --git a/modules/core/vehicle-identity/pom.xml b/modules/core/vehicle-identity/pom.xml index 3bc5f43d..723d1a35 100644 --- a/modules/core/vehicle-identity/pom.xml +++ b/modules/core/vehicle-identity/pom.xml @@ -28,6 +28,11 @@ com.fasterxml.jackson.core jackson-databind + + com.mysql + mysql-connector-j + runtime + org.junit.jupiter junit-jupiter diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/FileVehicleIdentityService.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/FileVehicleIdentityService.java index 119d5f1b..f15c2134 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/FileVehicleIdentityService.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/FileVehicleIdentityService.java @@ -40,6 +40,9 @@ public final class FileVehicleIdentityService implements VehicleIdentityResolver @Override public void bind(VehicleIdentityBinding binding) { + if (binding == null || !binding.hasResolvedVin()) { + return; + } delegate.bind(binding); synchronized (writeLock) { try { diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/InMemoryVehicleIdentityService.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/InMemoryVehicleIdentityService.java index fa71a2da..ca021d8f 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/InMemoryVehicleIdentityService.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/InMemoryVehicleIdentityService.java @@ -14,6 +14,9 @@ public final class InMemoryVehicleIdentityService implements VehicleIdentityReso @Override public void bind(VehicleIdentityBinding binding) { + if (binding == null || !binding.hasResolvedVin()) { + return; + } if (!binding.phone().isBlank()) { phoneToVin.put(key(binding.protocol(), binding.phone()), binding.vin()); } diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java new file mode 100644 index 00000000..66a2da5a --- /dev/null +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java @@ -0,0 +1,145 @@ +package com.lingniu.ingest.identity; + +import com.lingniu.ingest.api.ProtocolId; + +import javax.sql.DataSource; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Locale; + +public final class MySqlVehicleIdentityService implements VehicleIdentityResolver, VehicleIdentityRegistry { + + private final DataSource dataSource; + private final String table; + + public MySqlVehicleIdentityService(DataSource dataSource, String table) { + if (dataSource == null) { + throw new IllegalArgumentException("dataSource must not be null"); + } + this.dataSource = dataSource; + this.table = sanitizeTable(table); + initializeSchema(); + } + + @Override + public void bind(VehicleIdentityBinding binding) { + if (binding == null || !binding.hasResolvedVin()) { + return; + } + upsert(binding.protocol(), "PHONE", binding.phone(), binding.vin()); + upsert(binding.protocol(), "DEVICE_ID", binding.deviceId(), binding.vin()); + upsert(binding.protocol(), "PLATE", binding.plate(), binding.vin()); + } + + @Override + public VehicleIdentity resolve(VehicleIdentityLookup lookup) { + if (lookup == null) { + return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN); + } + if (!lookup.vin().isBlank()) { + return new VehicleIdentity(lookup.vin(), true, VehicleIdentitySource.EXPLICIT_VIN); + } + VehicleIdentity phone = resolveBound(lookup.protocol(), "PHONE", lookup.phone(), VehicleIdentitySource.BOUND_PHONE); + if (phone != null) return phone; + VehicleIdentity device = resolveBound(lookup.protocol(), "DEVICE_ID", lookup.deviceId(), VehicleIdentitySource.BOUND_DEVICE_ID); + if (device != null) return device; + VehicleIdentity plate = resolveBound(lookup.protocol(), "PLATE", lookup.plate(), VehicleIdentitySource.BOUND_PLATE); + if (plate != null) return plate; + if (!lookup.deviceId().isBlank()) { + return new VehicleIdentity(lookup.deviceId(), false, VehicleIdentitySource.FALLBACK_DEVICE_ID); + } + if (!lookup.phone().isBlank()) { + return new VehicleIdentity(lookup.phone(), false, VehicleIdentitySource.FALLBACK_PHONE); + } + if (!lookup.plate().isBlank()) { + return new VehicleIdentity(lookup.plate(), false, VehicleIdentitySource.FALLBACK_PLATE); + } + return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN); + } + + private void initializeSchema() { + String sql = """ + CREATE TABLE IF NOT EXISTS %s ( + protocol VARCHAR(32) NOT NULL, + identifier_type VARCHAR(32) NOT NULL, + identifier_value VARCHAR(128) NOT NULL, + vin VARCHAR(64) NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (protocol, identifier_type, identifier_value), + KEY idx_vehicle_identity_vin (vin) + ) + """.formatted(table); + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute(sql); + } catch (SQLException e) { + throw new IllegalStateException("vehicle identity mysql schema initialize failed: " + table, e); + } + } + + private void upsert(ProtocolId protocol, String identifierType, String identifierValue, String vin) { + String normalizedIdentifier = normalize(identifierValue); + if (normalizedIdentifier.isBlank()) { + return; + } + String sql = "INSERT INTO " + table + " (protocol, identifier_type, identifier_value, vin) " + + "VALUES (?, ?, ?, ?) " + + "ON DUPLICATE KEY UPDATE vin = VALUES(vin), updated_at = CURRENT_TIMESTAMP"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, protocolName(protocol)); + statement.setString(2, identifierType); + statement.setString(3, normalizedIdentifier); + statement.setString(4, vin.trim()); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("vehicle identity mysql bind failed: " + table, e); + } + } + + private VehicleIdentity resolveBound(ProtocolId protocol, + String identifierType, + String identifierValue, + VehicleIdentitySource source) { + String normalizedIdentifier = normalize(identifierValue); + if (normalizedIdentifier.isBlank()) { + return null; + } + String sql = "SELECT vin FROM " + table + + " WHERE protocol = ? AND identifier_type = ? AND identifier_value = ? LIMIT 1"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, protocolName(protocol)); + statement.setString(2, identifierType); + statement.setString(3, normalizedIdentifier); + try (ResultSet resultSet = statement.executeQuery()) { + if (resultSet.next()) { + return new VehicleIdentity(resultSet.getString(1), true, source); + } + } + } catch (SQLException e) { + throw new IllegalStateException("vehicle identity mysql resolve failed: " + table, e); + } + return null; + } + + private static String sanitizeTable(String table) { + String value = table == null || table.isBlank() ? "vehicle_identity_binding" : table.trim(); + if (!value.matches("[A-Za-z0-9_]+")) { + throw new IllegalArgumentException("vehicle identity mysql table must contain only letters, digits, and underscore"); + } + return value; + } + + private static String protocolName(ProtocolId protocol) { + return protocol == null ? "UNKNOWN" : protocol.name(); + } + + private static String normalize(String value) { + return value == null ? "" : value.trim().toUpperCase(Locale.ROOT); + } +} diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityBinding.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityBinding.java index d393dbe6..f5b8f97c 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityBinding.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityBinding.java @@ -19,6 +19,10 @@ public record VehicleIdentityBinding( } } + public boolean hasResolvedVin() { + return !"unknown".equalsIgnoreCase(vin); + } + private static String normalize(String value) { return value == null ? "" : value.trim(); } diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfiguration.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfiguration.java index 94ab994c..fb5d858c 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfiguration.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfiguration.java @@ -3,6 +3,7 @@ package com.lingniu.ingest.identity.config; import com.fasterxml.jackson.databind.ObjectMapper; import com.lingniu.ingest.identity.FileVehicleIdentityService; import com.lingniu.ingest.identity.InMemoryVehicleIdentityService; +import com.lingniu.ingest.identity.MySqlVehicleIdentityService; import com.lingniu.ingest.identity.VehicleIdentityRegistry; import com.lingniu.ingest.identity.VehicleIdentityResolver; import org.springframework.boot.autoconfigure.AutoConfiguration; @@ -11,7 +12,14 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; +import javax.sql.DataSource; +import java.io.PrintWriter; import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.util.logging.Logger; @AutoConfiguration @EnableConfigurationProperties(VehicleIdentityProperties.class) @@ -32,6 +40,23 @@ public class VehicleIdentityAutoConfiguration { return new FileVehicleIdentityService(Path.of(properties.getFile().getPath()), objectMapper); } + @Bean + @ConditionalOnMissingBean(DataSource.class) + @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "mysql") + public DataSource vehicleIdentityDataSource(VehicleIdentityProperties properties) { + VehicleIdentityProperties.Mysql mysql = properties.getMysql(); + return new DriverManagerVehicleIdentityDataSource( + mysql.getDriverClassName(), mysql.getJdbcUrl(), mysql.getUsername(), mysql.getPassword()); + } + + @Bean + @ConditionalOnMissingBean({VehicleIdentityResolver.class, VehicleIdentityRegistry.class}) + @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "mysql") + public MySqlVehicleIdentityService mySqlVehicleIdentityService(VehicleIdentityProperties properties, + DataSource dataSource) { + return new MySqlVehicleIdentityService(dataSource, properties.getMysql().getTable()); + } + @Bean @ConditionalOnMissingBean({VehicleIdentityResolver.class, VehicleIdentityRegistry.class}) @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "memory", @@ -39,4 +64,61 @@ public class VehicleIdentityAutoConfiguration { public InMemoryVehicleIdentityService vehicleIdentityService() { return new InMemoryVehicleIdentityService(); } + + private static final class DriverManagerVehicleIdentityDataSource implements DataSource { + private final String jdbcUrl; + private final String username; + private final String password; + private PrintWriter logWriter; + private int loginTimeout; + + private DriverManagerVehicleIdentityDataSource(String driverClassName, + String jdbcUrl, + String username, + String password) { + if (jdbcUrl == null || jdbcUrl.isBlank()) { + throw new IllegalArgumentException("lingniu.ingest.identity.mysql.jdbc-url is required"); + } + if (driverClassName != null && !driverClassName.isBlank()) { + try { + Class.forName(driverClassName.trim()); + } catch (ClassNotFoundException e) { + throw new IllegalArgumentException("vehicle identity mysql driver class not found: " + + driverClassName, e); + } + } + this.jdbcUrl = jdbcUrl; + this.username = username == null ? "" : username; + this.password = password == null ? "" : password; + } + + @Override + public Connection getConnection() throws SQLException { + DriverManager.setLoginTimeout(loginTimeout); + return DriverManager.getConnection(jdbcUrl, username, password); + } + + @Override + public Connection getConnection(String username, String password) throws SQLException { + DriverManager.setLoginTimeout(loginTimeout); + return DriverManager.getConnection(jdbcUrl, username, password); + } + + @Override public PrintWriter getLogWriter() { return logWriter; } + @Override public void setLogWriter(PrintWriter out) { this.logWriter = out; } + @Override public void setLoginTimeout(int seconds) { this.loginTimeout = seconds; } + @Override public int getLoginTimeout() { return loginTimeout; } + @Override public Logger getParentLogger() throws SQLFeatureNotSupportedException { + throw new SQLFeatureNotSupportedException(); + } + @Override public T unwrap(Class iface) throws SQLException { + if (iface.isInstance(this)) { + return iface.cast(this); + } + throw new SQLException("not a wrapper for " + iface); + } + @Override public boolean isWrapperFor(Class iface) { + return iface.isInstance(this); + } + } } diff --git a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityProperties.java b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityProperties.java index 50acf4b0..146efe74 100644 --- a/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityProperties.java +++ b/modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityProperties.java @@ -7,11 +7,14 @@ public class VehicleIdentityProperties { private String store = "memory"; private File file = new File(); + private Mysql mysql = new Mysql(); public String getStore() { return store; } public void setStore(String store) { this.store = store; } public File getFile() { return file; } public void setFile(File file) { this.file = file; } + public Mysql getMysql() { return mysql; } + public void setMysql(Mysql mysql) { this.mysql = mysql; } public static class File { private String path = "./data/vehicle-identity.jsonl"; @@ -19,4 +22,23 @@ public class VehicleIdentityProperties { public String getPath() { return path; } public void setPath(String path) { this.path = path; } } + + public static class Mysql { + private String table = "vehicle_identity_binding"; + private String jdbcUrl = ""; + private String username = ""; + private String password = ""; + private String driverClassName = "com.mysql.cj.jdbc.Driver"; + + public String getTable() { return table; } + public void setTable(String table) { this.table = table; } + public String getJdbcUrl() { return jdbcUrl; } + public void setJdbcUrl(String jdbcUrl) { this.jdbcUrl = jdbcUrl; } + public String getUsername() { return username; } + public void setUsername(String username) { this.username = username; } + public String getPassword() { return password; } + public void setPassword(String password) { this.password = password; } + public String getDriverClassName() { return driverClassName; } + public void setDriverClassName(String driverClassName) { this.driverClassName = driverClassName; } + } } diff --git a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java new file mode 100644 index 00000000..a5ac95b2 --- /dev/null +++ b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java @@ -0,0 +1,199 @@ +package com.lingniu.ingest.identity; + +import com.lingniu.ingest.api.ProtocolId; +import org.junit.jupiter.api.Test; + +import javax.sql.DataSource; +import java.io.PrintWriter; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.logging.Logger; + +import static org.assertj.core.api.Assertions.assertThat; + +class MySqlVehicleIdentityServiceTest { + + @Test + void bindsAndResolvesIdentityThroughJdbcStore() { + RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); + MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); + + service.bind(new VehicleIdentityBinding( + ProtocolId.JT808, "LNVIN000000000123", "13079960001", "DEV001", "粤B00001")); + + VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "13079960001", "", "")); + VehicleIdentity byDevice = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "DEV001", "")); + VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "", "", "粤B00001")); + + assertThat(jdbc.createdTables).singleElement() + .satisfies(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding")); + assertThat(jdbc.bindings).containsKey("JT808|PHONE|13079960001"); + assertThat(byPhone.vin()).isEqualTo("LNVIN000000000123"); + assertThat(byPhone.resolved()).isTrue(); + assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE); + assertThat(byDevice.vin()).isEqualTo("LNVIN000000000123"); + assertThat(byDevice.source()).isEqualTo(VehicleIdentitySource.BOUND_DEVICE_ID); + assertThat(byPlate.vin()).isEqualTo("LNVIN000000000123"); + assertThat(byPlate.source()).isEqualTo(VehicleIdentitySource.BOUND_PLATE); + } + + @Test + void ignoresUnknownVinBinding() { + RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc(); + MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding"); + + service.bind(new VehicleIdentityBinding( + ProtocolId.JT808, "unknown", "13079960001", "DEV001", "粤B00001")); + + assertThat(jdbc.bindings).isEmpty(); + } + + private static final class RecordingIdentityJdbc { + private final List createdTables = new ArrayList<>(); + private final Map bindings = new HashMap<>(); + + private DataSource dataSource() { + return new DataSource() { + @Override public Connection getConnection() { return connection(); } + @Override public Connection getConnection(String username, String password) { return connection(); } + @Override public PrintWriter getLogWriter() { return null; } + @Override public void setLogWriter(PrintWriter out) {} + @Override public void setLoginTimeout(int seconds) {} + @Override public int getLoginTimeout() { return 0; } + @Override public Logger getParentLogger() throws SQLFeatureNotSupportedException { throw new SQLFeatureNotSupportedException(); } + @Override public T unwrap(Class iface) { throw new UnsupportedOperationException(); } + @Override public boolean isWrapperFor(Class iface) { return false; } + }; + } + + private Connection connection() { + InvocationHandler handler = (proxy, method, args) -> switch (method.getName()) { + case "createStatement" -> statement(); + case "prepareStatement" -> preparedStatement((String) args[0]); + case "close" -> null; + case "isClosed" -> false; + case "unwrap" -> throw new UnsupportedOperationException(); + case "isWrapperFor" -> false; + default -> defaultValue(method.getReturnType()); + }; + return (Connection) Proxy.newProxyInstance( + Connection.class.getClassLoader(), new Class[]{Connection.class}, handler); + } + + private Statement statement() { + InvocationHandler handler = (proxy, method, args) -> { + if ("execute".equals(method.getName())) { + createdTables.add((String) args[0]); + return false; + } + if ("close".equals(method.getName())) { + return null; + } + return defaultValue(method.getReturnType()); + }; + return (Statement) Proxy.newProxyInstance( + Statement.class.getClassLoader(), new Class[]{Statement.class}, handler); + } + + private PreparedStatement preparedStatement(String sql) { + List params = new ArrayList<>(); + InvocationHandler handler = (proxy, method, args) -> { + switch (method.getName()) { + case "setString" -> { + set(params, (Integer) args[0], args[1]); + return null; + } + case "executeUpdate" -> { + upsert(params); + return 1; + } + case "executeQuery" -> { + return resultSet(resolve(params)); + } + case "close" -> { + return null; + } + default -> { + return defaultValue(method.getReturnType()); + } + } + }; + return (PreparedStatement) Proxy.newProxyInstance( + PreparedStatement.class.getClassLoader(), new Class[]{PreparedStatement.class}, handler); + } + + private void upsert(List params) { + String protocol = (String) params.get(0); + String identifierType = (String) params.get(1); + String identifierValue = normalize((String) params.get(2)); + String vin = (String) params.get(3); + bindings.put(protocol + "|" + identifierType + "|" + identifierValue, vin); + } + + private String resolve(List params) { + String protocol = (String) params.get(0); + String identifierType = (String) params.get(1); + String identifierValue = normalize((String) params.get(2)); + return bindings.get(protocol + "|" + identifierType + "|" + identifierValue); + } + + private static ResultSet resultSet(String vin) { + InvocationHandler handler = new InvocationHandler() { + private boolean read; + + @Override + public Object invoke(Object proxy, java.lang.reflect.Method method, Object[] args) { + return switch (method.getName()) { + case "next" -> { + boolean hasNext = !read && vin != null; + read = true; + yield hasNext; + } + case "getString" -> vin; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }; + } + }; + return (ResultSet) Proxy.newProxyInstance( + ResultSet.class.getClassLoader(), new Class[]{ResultSet.class}, handler); + } + + private static void set(List params, int oneBasedIndex, Object value) { + while (params.size() < oneBasedIndex) { + params.add(null); + } + params.set(oneBasedIndex - 1, value); + } + + private static String normalize(String value) { + return value == null ? "" : value.trim().toUpperCase(Locale.ROOT); + } + + private static Object defaultValue(Class type) { + if (type == Void.TYPE) return null; + if (type == Boolean.TYPE) return false; + if (type == Integer.TYPE) return 0; + if (type == Long.TYPE) return 0L; + if (type == Double.TYPE) return 0D; + if (type == Float.TYPE) return 0F; + if (type == Short.TYPE) return (short) 0; + if (type == Byte.TYPE) return (byte) 0; + return null; + } + } +} diff --git a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java index 298bb862..1bc8a183 100644 --- a/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java +++ b/modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java @@ -7,7 +7,12 @@ import org.junit.jupiter.api.io.TempDir; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import javax.sql.DataSource; import java.nio.file.Path; +import java.sql.Connection; +import java.sql.Statement; +import java.sql.SQLException; +import java.lang.reflect.Proxy; import static org.assertj.core.api.Assertions.assertThat; @@ -41,4 +46,59 @@ class VehicleIdentityAutoConfigurationTest { .isEqualTo("FileVehicleIdentityService"); }); } + + @Test + void createsMysqlIdentityServiceWhenConfigured() { + runner + .withBean(DataSource.class, StubDataSource::new) + .withPropertyValues( + "lingniu.ingest.identity.store=mysql", + "lingniu.ingest.identity.mysql.table=vehicle_identity_binding") + .run(context -> { + assertThat(context).hasSingleBean(VehicleIdentityResolver.class); + assertThat(context.getBean(VehicleIdentityResolver.class).getClass().getSimpleName()) + .isEqualTo("MySqlVehicleIdentityService"); + assertThat(context.getBean(VehicleIdentityResolver.class)) + .isSameAs(context.getBean(VehicleIdentityRegistry.class)); + }); + } + + private static final class StubDataSource implements DataSource { + @Override public Connection getConnection() { return connection(); } + @Override public Connection getConnection(String username, String password) { return connection(); } + @Override public java.io.PrintWriter getLogWriter() { return null; } + @Override public void setLogWriter(java.io.PrintWriter out) {} + @Override public void setLoginTimeout(int seconds) {} + @Override public int getLoginTimeout() { return 0; } + @Override public java.util.logging.Logger getParentLogger() throws java.sql.SQLFeatureNotSupportedException { throw new java.sql.SQLFeatureNotSupportedException(); } + @Override public T unwrap(Class iface) { throw new UnsupportedOperationException(); } + @Override public boolean isWrapperFor(Class iface) { return false; } + + private static Connection connection() { + return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), new Class[]{Connection.class}, + (proxy, method, args) -> switch (method.getName()) { + case "createStatement" -> statement(); + case "close" -> null; + case "isClosed" -> false; + default -> defaultValue(method.getReturnType()); + }); + } + + private static Statement statement() { + return (Statement) Proxy.newProxyInstance(Statement.class.getClassLoader(), new Class[]{Statement.class}, + (proxy, method, args) -> switch (method.getName()) { + case "execute" -> false; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }); + } + + private static Object defaultValue(Class type) { + if (type == Void.TYPE) return null; + if (type == Boolean.TYPE) return false; + if (type == Integer.TYPE) return 0; + if (type == Long.TYPE) return 0L; + return null; + } + } } diff --git a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java index 294dbc85..da88dc8d 100644 --- a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java +++ b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java @@ -287,7 +287,9 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { Map.of("protocolVersion", msg.header().version().name()))); session = session.withVin(identity.vin()); if (msg.body() instanceof Jt808Body.Register reg && reg.plate() != null) { - if (session.vin() != null && !session.vin().isBlank()) { + if (identity.resolved() && session.vin() != null + && !session.vin().isBlank() + && !"unknown".equalsIgnoreCase(session.vin())) { identityRegistry.bind(new VehicleIdentityBinding( ProtocolId.JT808, session.vin(), phone, reg.deviceId(), reg.plate())); } diff --git a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java index 2fc838c0..cb774bc4 100644 --- a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java +++ b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java @@ -14,6 +14,8 @@ import com.lingniu.ingest.core.dispatcher.HandlerRegistry; import com.lingniu.ingest.core.pipeline.InterceptorChain; import com.lingniu.ingest.identity.InMemoryVehicleIdentityService; import com.lingniu.ingest.identity.VehicleIdentityBinding; +import com.lingniu.ingest.identity.VehicleIdentityLookup; +import com.lingniu.ingest.identity.VehicleIdentitySource; import com.lingniu.ingest.protocol.jt808.codec.BodyParserRegistry; import com.lingniu.ingest.protocol.jt808.codec.Jt808MalformedFrame; import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder; @@ -82,6 +84,45 @@ class Jt808ChannelHandlerTest { eventBus.close(); } + @Test + void unresolvedRegisterDoesNotBindUnknownVinToExternalIdentifiers() { + InMemorySessionStore sessions = new InMemorySessionStore(); + InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); + Dispatcher dispatcher = new Dispatcher( + new HandlerRegistry(), + new InterceptorChain(List.of()), + new HandlerInvoker(), + eventBus, + batchExecutor); + Jt808ChannelHandler handler = new Jt808ChannelHandler( + new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))), + dispatcher, + sessions, + identity, + new Jt808ChannelRegistry(), + new Jt808PendingRequests()); + EmbeddedChannel channel = new EmbeddedChannel(handler); + + channel.writeInbound(buildFrame( + Jt808MessageId.TERMINAL_REGISTER, + "123456789012", + 1, + buildRegisterBody("DEV808", "B80808"))); + + assertThat(identity.resolve(new VehicleIdentityLookup( + ProtocolId.JT808, "", "123456789012", "", ""))) + .satisfies(resolved -> { + assertThat(resolved.vin()).isEqualTo("123456789012"); + assertThat(resolved.resolved()).isFalse(); + assertThat(resolved.source()).isEqualTo(VehicleIdentitySource.FALLBACK_PHONE); + }); + + batchExecutor.close(); + eventBus.close(); + } + @Test void authSessionUsesResolvedInternalVinFromImei() { InMemorySessionStore sessions = new InMemorySessionStore();