refactor: require mqtt identity resolver

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<MqttAsyncClient> 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;
     }
 

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<MqttPayload> {
 
     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

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();

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() {

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() {