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

refactor: require explicit history consumer bindings

Browse Source
This commit is contained in:
lingniu
2026-07-01 05:00:39 +08:00
parent b841141fed
commit 6016e3f99c
3 changed files with 59 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java
View File

@@ -27,11 +27,9 @@ public class VehicleHistoryKafkaConsumerConfiguration {
SinkMqProperties props) {
List<KafkaEnvelopeConsumerWorker> workers = new KafkaEnvelopeConsumerFactory().createWorkers(
Map.of(
"eventHistoryEnvelopeConsumerProcessor", processor,
"eventHistoryGb32960EnvelopeConsumerProcessor", processor,
"eventHistoryJt808EnvelopeConsumerProcessor", processor,
"eventHistoryYutongMqttEnvelopeConsumerProcessor", processor,
"eventHistoryRawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryGb32960RawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryJt808RawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryYutongMqttRawEnvelopeConsumerProcessor", rawProcessor),

14
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java
View File

@@ -29,6 +29,20 @@ class MavenModuleProfileTest {
"modules/apps/command-gateway");
}
@Test
void xindaIsLegacyOnlyAndOutsideDefaultProductionReactor() throws Exception {
Document pom = rootPom();
assertThat(defaultModules(pom))
.doesNotContain("modules/inbound/inbound-xinda-push")
.doesNotContain("modules/apps/xinda-push-app");
assertThat(profileModules(pom, "legacy-xinda"))
.containsExactly(
"modules/inbound/inbound-xinda-push",
"modules/apps/xinda-push-app");
}
private static Document rootPom() throws Exception {
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);

45
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java
View File

@@ -1,6 +1,8 @@
package com.lingniu.ingest.historyapp;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
@@ -18,6 +20,7 @@ import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
import com.lingniu.ingest.sink.mq.KafkaEventSink;
import com.lingniu.ingest.sink.mq.SinkMqProperties;
import com.lingniu.ingest.sink.mq.proto.ParseStatusProto;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload;
@@ -39,6 +42,7 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
@@ -65,6 +69,23 @@ class VehicleHistoryAppCompositionTest {
.doesNotContain("eventHistoryRawEnvelopeConsumerProcessor");
}
@Test
void kafkaConsumerRunnerRejectsGenericHistoryBindings() {
new ApplicationContextRunner()
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
.withBean("eventHistoryEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class,
() -> processor("event-history"))
.withBean("eventHistoryRawEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class,
() -> processor("event-history-raw"))
.withBean(SinkMqProperties.class, VehicleHistoryAppCompositionTest::genericOnlyHistoryConsumerProps)
.withPropertyValues("lingniu.ingest.sink.mq.consumer.enabled=true")
.run(context -> {
assertThat(context).hasFailed();
assertThat(context.getStartupFailure())
.hasRootCauseMessage("no vehicle history kafka consumer workers created; check consumer bindings");
});
}
@Test
void createsTdengineHistoryStorageAndQueryBeansWithoutGb32960TcpServerOrEventFileStore() {
new ApplicationContextRunner()
@@ -160,6 +181,30 @@ class VehicleHistoryAppCompositionTest {
return mock(KafkaProducer.class);
}
private static EnvelopeConsumerProcessor processor(String service) {
return new EnvelopeConsumerProcessor(
service,
ignored -> EnvelopeIngestResult.processed("event-1", "VIN001"),
ignored -> {});
}
private static SinkMqProperties genericOnlyHistoryConsumerProps() {
SinkMqProperties props = new SinkMqProperties();
props.setBootstrapServers("localhost:9092");
props.getConsumer().setEnabled(true);
props.getConsumer().setBindings(Map.of(
"eventHistoryEnvelopeConsumerProcessor", binding("history-event", "vehicle.event.generic.v1"),
"eventHistoryRawEnvelopeConsumerProcessor", binding("history-raw", "vehicle.raw.generic.v1")));
return props;
}
private static SinkMqProperties.Binding binding(String groupId, String topic) {
SinkMqProperties.Binding binding = new SinkMqProperties.Binding();
binding.setGroupId(groupId);
binding.setTopics(List.of(topic));
return binding;
}
private static VehicleEnvelope rawFrameEnvelope() {
return VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")