diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java index 61a1ff51..c710e8ce 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java @@ -3,7 +3,6 @@ package com.lingniu.ingest.inbound.mqtt.client; import com.lingniu.ingest.api.ProtocolId; import com.lingniu.ingest.api.pipeline.RawFrame; import com.lingniu.ingest.core.dispatcher.Dispatcher; -import com.lingniu.ingest.identity.StatelessVehicleIdentityResolver; import com.lingniu.ingest.identity.VehicleIdentity; import com.lingniu.ingest.identity.VehicleIdentityLookup; import com.lingniu.ingest.identity.VehicleIdentityResolver; @@ -50,19 +49,16 @@ public final class MqttEndpointManager implements AutoCloseable { private final List clients = new ArrayList<>(); private boolean started; - public MqttEndpointManager(MqttInboundProperties props, - MqttProfileRegistry profileRegistry, - Dispatcher dispatcher) { - this(props, profileRegistry, new StatelessVehicleIdentityResolver(), dispatcher); - } - public MqttEndpointManager(MqttInboundProperties props, MqttProfileRegistry profileRegistry, VehicleIdentityResolver identityResolver, Dispatcher dispatcher) { this.props = props; this.profileRegistry = profileRegistry; - this.identityResolver = identityResolver == null ? new StatelessVehicleIdentityResolver() : identityResolver; + if (identityResolver == null) { + throw new IllegalArgumentException("identityResolver must not be null"); + } + this.identityResolver = identityResolver; this.dispatcher = dispatcher; } diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java index 3087a925..e901ea38 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java @@ -6,7 +6,6 @@ import com.lingniu.ingest.api.event.LocationPayload; import com.lingniu.ingest.api.event.RealtimePayload; import com.lingniu.ingest.api.event.VehicleEvent; import com.lingniu.ingest.api.spi.EventMapper; -import com.lingniu.ingest.identity.StatelessVehicleIdentityResolver; import com.lingniu.ingest.identity.VehicleIdentity; import com.lingniu.ingest.identity.VehicleIdentityLookup; import com.lingniu.ingest.identity.VehicleIdentityResolver; @@ -30,12 +29,11 @@ public final class YutongEventMapper implements EventMapper { private final VehicleIdentityResolver identityResolver; - public YutongEventMapper() { - this(new StatelessVehicleIdentityResolver()); - } - public YutongEventMapper(VehicleIdentityResolver identityResolver) { - this.identityResolver = identityResolver == null ? new StatelessVehicleIdentityResolver() : identityResolver; + if (identityResolver == null) { + throw new IllegalArgumentException("identityResolver must not be null"); + } + this.identityResolver = identityResolver; } @Override diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java index 138c7392..4efa29d2 100644 --- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java +++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java @@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class MqttEndpointManagerTest { @@ -98,7 +99,7 @@ class MqttEndpointManagerTest { MqttProfileRegistry profileRegistry = new MqttProfileRegistry(List.of( new YutongMqttProfile( new MqttPayloadParser(new ObjectMapper()), - new YutongEventMapper()))); + new YutongEventMapper(new InMemoryVehicleIdentityService())))); Dispatcher dispatcher = new Dispatcher( registry(profileRegistry), new InterceptorChain(List.of()), @@ -106,7 +107,7 @@ class MqttEndpointManagerTest { eventBus, batchExecutor); MqttEndpointManager manager = new MqttEndpointManager( - new MqttInboundProperties(), profileRegistry, dispatcher); + new MqttInboundProperties(), profileRegistry, new InMemoryVehicleIdentityService(), dispatcher); invokeOnMessage(manager, endpoint, "/ytforward/bad", badPayload); @@ -159,7 +160,8 @@ class MqttEndpointManagerTest { new HandlerInvoker(), eventBus, batchExecutor); - MqttEndpointManager manager = new MqttEndpointManager(props, profileRegistry, dispatcher); + MqttEndpointManager manager = new MqttEndpointManager( + props, profileRegistry, new InMemoryVehicleIdentityService(), dispatcher); manager.start(); @@ -209,7 +211,7 @@ class MqttEndpointManagerTest { eventBus, batchExecutor); MqttEndpointManager manager = new MqttEndpointManager( - new MqttInboundProperties(), profileRegistry, dispatcher); + new MqttInboundProperties(), profileRegistry, new InMemoryVehicleIdentityService(), dispatcher); invokeOnMessage(manager, endpoint, "/vehicles/broken", payload); @@ -245,6 +247,27 @@ class MqttEndpointManagerTest { eventBus.close(); } + @Test + void rejectsMissingIdentityResolver() throws Exception { + MqttProfileRegistry profileRegistry = new MqttProfileRegistry(List.of()); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); + Dispatcher dispatcher = new Dispatcher( + registry(profileRegistry), + new InterceptorChain(List.of()), + new HandlerInvoker(), + eventBus, + batchExecutor); + + assertThatThrownBy(() -> new MqttEndpointManager( + new MqttInboundProperties(), profileRegistry, null, dispatcher)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("identityResolver"); + + batchExecutor.close(); + eventBus.close(); + } + @Test void identityResolverFailureStillArchivesAndDispatchesRealtimeWithUnknownIdentity() throws Exception { MqttInboundProperties.Endpoint endpoint = new MqttInboundProperties.Endpoint(); diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttProfileRegistryTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttProfileRegistryTest.java index f4e58f7b..c6d8add3 100644 --- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttProfileRegistryTest.java +++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttProfileRegistryTest.java @@ -2,6 +2,7 @@ package com.lingniu.ingest.inbound.mqtt; import com.fasterxml.jackson.databind.ObjectMapper; import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.identity.InMemoryVehicleIdentityService; import com.lingniu.ingest.inbound.mqtt.mapper.YutongEventMapper; import com.lingniu.ingest.inbound.mqtt.parser.MqttPayloadParser; import com.lingniu.ingest.inbound.mqtt.profile.MqttProfileRegistry; @@ -16,7 +17,8 @@ import static org.assertj.core.api.Assertions.assertThat; class MqttProfileRegistryTest { private final MqttProfileRegistry registry = new MqttProfileRegistry(List.of( - new YutongMqttProfile(new MqttPayloadParser(new ObjectMapper()), new YutongEventMapper()))); + new YutongMqttProfile(new MqttPayloadParser(new ObjectMapper()), + new YutongEventMapper(new InMemoryVehicleIdentityService())))); @Test void resolvesProfileCaseInsensitiveAndRoutesParseAndMapping() { diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/YutongEventMapperTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/YutongEventMapperTest.java index 9677e82e..3aec762e 100644 --- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/YutongEventMapperTest.java +++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/YutongEventMapperTest.java @@ -13,12 +13,20 @@ import org.junit.jupiter.api.Test; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class YutongEventMapperTest { private final ObjectMapper objectMapper = new ObjectMapper(); private final MqttPayloadParser parser = new MqttPayloadParser(objectMapper); - private final YutongEventMapper mapper = new YutongEventMapper(); + private final YutongEventMapper mapper = new YutongEventMapper(new InMemoryVehicleIdentityService()); + + @Test + void rejectsMissingIdentityResolver() { + assertThatThrownBy(() -> new YutongEventMapper(null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("identityResolver"); + } @Test void parsesYutongSamplePayload() {