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

feat: route gb32960 kafka topics

Browse Source
This commit is contained in:
lingniu
2026-06-23 16:50:07 +08:00
parent 8e9afd79ac
commit 66807fdf3d
4 changed files with 196 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java
View File

@@ -43,13 +43,11 @@ public final class KafkaEventSink implements EventSink, AutoCloseable {
}
/**
* 拒绝 {@link VehicleEvent.RawArchive} —— 当前生产 Kafka sink 只投递实时、会话、告警等轻量事件。
* {@link EnvelopeMapper} 和 {@link TopicRouter} 已具备 RAW envelope/topic 的映射能力，但这里仍
* 显式过滤；如需把 32960 raw 引用转发到 {@code vehicle.raw.archive}，应先移除此过滤并补充测试。
* Split-service mode uses Kafka for both normalized events and raw archive records.
*/
@Override
public boolean accepts(VehicleEvent event) {
return !(event instanceof VehicleEvent.RawArchive);
return true;
}
@Override

14
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java
View File

@@ -57,16 +57,16 @@ public class SinkMqProperties {
public static class Topics {
/** 实时遥测事件 topic；GB32960 RAW-only 架构下可逐步弱化该 topic。 */
private String realtime = "vehicle.realtime";
private String realtime = "vehicle.event.gb32960.v1";
/** 位置事件 topic；通常可由 realtime/RAW 派生。 */
private String location = "vehicle.location";
private String alarm = "vehicle.alarm";
private String session = "vehicle.session";
private String mediaMeta = "vehicle.media.meta";
private String location = "vehicle.event.gb32960.v1";
private String alarm = "vehicle.event.gb32960.v1";
private String session = "vehicle.event.gb32960.v1";
private String mediaMeta = "vehicle.media.meta.v1";
/** RAW 归档引用 topic，payload 应携带 archive URI/size，不建议携带完整 raw bytes。 */
private String rawArchive = "vehicle.raw.archive";
private String rawArchive = "vehicle.raw.gb32960.v1";
/** producer 熔断或 consumer 处理失败时的死信 topic。 */
private String dlq = "vehicle.dlq";
private String dlq = "vehicle.dlq.gb32960.v1";
public String getRealtime() { return realtime; }
public void setRealtime(String realtime) { this.realtime = realtime; }

40
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java Normal file
View File

@@ -0,0 +1,40 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.VehicleEvent;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
class KafkaEventSinkTest {
@Test
void acceptsRawArchiveRecordsForSplitServiceKafkaBoundary() {
KafkaEventSink sink = new KafkaEventSink(
null,
new EnvelopeMapper("node-1"),
new TopicRouter(new SinkMqProperties.Topics()),
"vehicle.dlq.gb32960.v1",
CircuitBreaker.ofDefaults("kafka-test"));
assertThat(sink.accepts(rawArchive())).isTrue();
}
private static VehicleEvent.RawArchive rawArchive() {
return new VehicleEvent.RawArchive(
"raw-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-raw-1",
Map.of("platformAccount", "Hyundai"),
0x02,
0,
new byte[] {0x23, 0x23});
}
}

147
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java Normal file
View File

@@ -0,0 +1,147 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.AlarmPayload;
import com.lingniu.ingest.api.event.LocationPayload;
import com.lingniu.ingest.api.event.RealtimePayload;
import com.lingniu.ingest.api.event.VehicleEvent;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
class TopicRouterTest {
@Test
void rawArchiveRoutesToVersionedGb32960RawTopicByDefault() {
TopicRouter router = new TopicRouter(new SinkMqProperties.Topics());
assertThat(router.route(rawArchive())).isEqualTo("vehicle.raw.gb32960.v1");
}
@Test
void normalizedGb32960EventsRouteToVersionedEventTopicByDefault() {
TopicRouter router = new TopicRouter(new SinkMqProperties.Topics());
assertThat(router.route(realtime())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(location())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(alarm())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(login())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(logout())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(heartbeat())).isEqualTo("vehicle.event.gb32960.v1");
}
@Test
void dlqDefaultsToVersionedGb32960DlqTopic() {
SinkMqProperties.Topics topics = new SinkMqProperties.Topics();
assertThat(topics.getDlq()).isEqualTo("vehicle.dlq.gb32960.v1");
}
private static VehicleEvent.RawArchive rawArchive() {
return new VehicleEvent.RawArchive(
"raw-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-raw-1",
Map.of("platformAccount", "Hyundai"),
0x02,
0,
new byte[] {0x23, 0x23});
}
private static VehicleEvent.Realtime realtime() {
return new VehicleEvent.Realtime(
"event-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-1",
Map.of("platformAccount", "Hyundai"),
new RealtimePayload(
42.1, 1234.5, 80.0, null, null,
null, null, null, null, null, null,
RealtimePayload.VehicleState.STARTED,
RealtimePayload.ChargingState.UNCHARGED,
RealtimePayload.RunningMode.ELECTRIC,
null, null, null,
113.12, 23.45, null, null, null, null));
}
private static VehicleEvent.Location location() {
return new VehicleEvent.Location(
"event-2",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-2",
Map.of("platformAccount", "Hyundai"),
new LocationPayload(113.12, 23.45, 0, 42.1, 90, 0, 0));
}
private static VehicleEvent.Alarm alarm() {
return new VehicleEvent.Alarm(
"event-3",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-3",
Map.of("platformAccount", "Hyundai"),
new AlarmPayload(
AlarmPayload.AlarmLevel.MINOR,
1,
"GB32960_ALARM",
List.of("1"),
Set.of("bit0"),
113.12,
23.45,
AlarmPayload.SafetyCategory.GENERAL,
false,
AlarmPayload.HydrogenLeakLevel.NONE,
false));
}
private static VehicleEvent.Login login() {
return new VehicleEvent.Login(
"event-4",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-4",
Map.of("platformAccount", "Hyundai"),
"iccid-1",
"V2016");
}
private static VehicleEvent.Logout logout() {
return new VehicleEvent.Logout(
"event-5",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-5",
Map.of("platformAccount", "Hyundai"));
}
private static VehicleEvent.Heartbeat heartbeat() {
return new VehicleEvent.Heartbeat(
"event-6",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-6",
Map.of("platformAccount", "Hyundai"));
}
}