feat: productionize mqtt ingress and trim history APIs

2026-06-30 01:28:51 +08:00
parent 1e2e59bddc
commit bf6728041b
15 changed files with 554 additions and 158 deletions
--- a/modules/apps/vehicle-history-app/src/main/resources/application.yml
+++ b/modules/apps/vehicle-history-app/src/main/resources/application.yml
@@ -114,6 +114,8 @@ lingniu:
      telemetry-fields-enabled: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:false}
    event-history:
      enabled: true
+      api:
+        specialized-enabled: ${EVENT_HISTORY_SPECIALIZED_API_ENABLED:false}
    vehicle-state:
      enabled: false
    vehicle-stat:
--- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java
+++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java
@@ -10,6 +10,7 @@ import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
 import com.lingniu.ingest.eventhistory.Gb32960FrameController;
 import com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration;
 import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
+import com.lingniu.ingest.eventhistory.LocationHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
 import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
 import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
@@ -81,9 +82,10 @@ class VehicleHistoryAppCompositionTest {
                    assertThat(context).hasSingleBean(EventHistoryController.class);
                    assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
                    assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
-                    assertThat(context).hasSingleBean(Gb32960FrameController.class);
-                    assertThat(context).hasSingleBean(Jt808LocationHistoryController.class);
+                    assertThat(context).hasSingleBean(LocationHistoryController.class);
                    assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class);
+                    assertThat(context).doesNotHaveBean(Gb32960FrameController.class);
+                    assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
                    assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
                });
    }
--- a/modules/inbound/inbound-mqtt/pom.xml
+++ b/modules/inbound/inbound-mqtt/pom.xml
@@ -9,7 +9,7 @@
    </parent>
    <artifactId>inbound-mqtt</artifactId>
    <name>inbound-mqtt</name>
-    <description>MQTT 接入模块（支持多 endpoint，宇通等），HiveMQ MQTT Client + 双向 TLS。</description>
+    <description>MQTT 接入模块（支持多 endpoint，宇通等），Fusesource MQTT Client + 双向 TLS。</description>

    <dependencies>
        <dependency>
@@ -34,8 +34,8 @@
            <optional>true</optional>
        </dependency>
        <dependency>
-            <groupId>com.hivemq</groupId>
-            <artifactId>hivemq-mqtt-client</artifactId>
+            <groupId>org.fusesource.mqtt-client</groupId>
+            <artifactId>mqtt-client</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
--- 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
@@ -1,9 +1,5 @@
 package com.lingniu.ingest.inbound.mqtt.client;

-import com.hivemq.client.mqtt.datatypes.MqttQos;
-import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
-import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
-import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
 import com.lingniu.ingest.api.ProtocolId;
 import com.lingniu.ingest.api.pipeline.RawFrame;
 import com.lingniu.ingest.core.dispatcher.Dispatcher;
@@ -15,27 +11,30 @@ import com.lingniu.ingest.identity.VehicleIdentitySource;
 import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
 import com.lingniu.ingest.inbound.mqtt.model.MqttPayload;
 import com.lingniu.ingest.inbound.mqtt.profile.MqttProfileRegistry;
+import org.fusesource.mqtt.client.Future;
+import org.fusesource.mqtt.client.FutureConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Message;
+import org.fusesource.mqtt.client.QoS;
+import org.fusesource.mqtt.client.Topic;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import java.net.URI;
+import java.io.EOFException;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;

 /**
 * MQTT 多 endpoint 生命周期管理。
 *
- * <p>为每个 endpoint 创建一个 HiveMQ 异步客户端 → 连接 → 订阅 → 注册回调 →
+ * <p>为每个 endpoint 创建一个 fusesource MQTT FutureConnection → 连接 → 订阅 → 接收循环 →
 * 回调里解析 payload → {@link Dispatcher#dispatch(RawFrame)}。
 *
- * <p>Netty EventLoop 回调不做业务处理，虚拟线程由 Dispatcher 侧的 Disruptor 承接。
+ * <p>每个 endpoint 在独立虚拟线程中阻塞接收，业务处理由 Dispatcher 侧的 Disruptor 承接。
 */
 public final class MqttEndpointManager implements AutoCloseable {

@@ -45,7 +44,9 @@ public final class MqttEndpointManager implements AutoCloseable {
    private final MqttProfileRegistry profileRegistry;
    private final VehicleIdentityResolver identityResolver;
    private final Dispatcher dispatcher;
-    private final List<Mqtt3AsyncClient> clients = new ArrayList<>();
+    private final List<FutureConnection> connections = new ArrayList<>();
+    private final List<Thread> workers = new ArrayList<>();
+    private volatile boolean running;

    public MqttEndpointManager(MqttInboundProperties props,
                               MqttProfileRegistry profileRegistry,
@@ -64,90 +65,110 @@ public final class MqttEndpointManager implements AutoCloseable {
    }

    public void start() {
+        running = true;
        log.info("mqtt endpoint manager starting, endpoints={}", props.getEndpoints().size());
        for (MqttInboundProperties.Endpoint ep : props.getEndpoints()) {
+            Thread worker = Thread.ofVirtual()
+                    .name("mqtt-endpoint-" + safeName(ep.getName()))
+                    .start(() -> runEndpoint(ep));
+            workers.add(worker);
+        }
+    }
+
+    private void runEndpoint(MqttInboundProperties.Endpoint ep) {
+        while (running && !Thread.currentThread().isInterrupted()) {
+            FutureConnection connection = null;
            try {
-                // 多 endpoint 互不影响：单个连接初始化失败会派发 operationalError，但不会阻止其他 endpoint。
                log.info("mqtt endpoint [{}] initializing uri={} topic={} qos={} tls={}",
                        ep.getName(), ep.getUri(), ep.getTopic(), ep.getQos(), hasCustomTls(ep.getTls()));
-                Mqtt3AsyncClient client = buildClient(ep);
-                log.info("mqtt endpoint [{}] client built", ep.getName());
-                clients.add(client);
-                connectAndSubscribe(client, ep);
-            } catch (Exception e) {
-                log.error("mqtt endpoint [{}] init failed, skipped", ep.getName(), e);
-                publishOperationalError(ep, "init", e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
-            }
-        }
-    }
-
-    private Mqtt3AsyncClient buildClient(MqttInboundProperties.Endpoint ep) {
-        URI uri = URI.create(ep.getUri());
-        int port = uri.getPort() > 0 ? uri.getPort() : defaultPort(uri.getScheme());
-        var builder = Mqtt3Client.builder()
-                .identifier(ep.getClientId() == null || ep.getClientId().isBlank()
-                        ? "lingniu-" + UUID.randomUUID() : ep.getClientId())
-                .serverHost(uri.getHost())
-                .serverPort(port)
-                .automaticReconnectWithDefaultConfig();
-        if (ep.getConnectionTimeoutSeconds() > 0) {
-            builder = builder.transportConfig()
-                    .serverHost(uri.getHost())
-                    .serverPort(port)
-                    .socketConnectTimeout(ep.getConnectionTimeoutSeconds(), TimeUnit.SECONDS)
-                    .mqttConnectTimeout(ep.getConnectionTimeoutSeconds(), TimeUnit.SECONDS)
-                    .applyTransportConfig();
-        }
-
-        if ("ssl".equalsIgnoreCase(uri.getScheme()) || "tls".equalsIgnoreCase(uri.getScheme())
-                || "mqtts".equalsIgnoreCase(uri.getScheme())) {
-            if (hasCustomTls(ep.getTls())) {
-                builder = builder.sslConfig(MqttTlsSupport.build(ep.getTls()));
-            } else {
-                builder = builder.sslWithDefaultConfig();
-            }
-        }
-        return builder.buildAsync();
-    }
-
-    private void connectAndSubscribe(Mqtt3AsyncClient client, MqttInboundProperties.Endpoint ep) {
-        var connectBuilder = client.connectWith()
-                .cleanSession(ep.isCleanSession());
-        if (ep.getKeepAliveSeconds() > 0) {
-            connectBuilder = connectBuilder.keepAlive(ep.getKeepAliveSeconds());
-        }
-        if (ep.getUsername() != null && !ep.getUsername().isBlank()) {
-            connectBuilder = connectBuilder.simpleAuth()
-                    .username(ep.getUsername())
-                    .password(ep.getPassword() == null ? new byte[0] : ep.getPassword().getBytes())
-                    .applySimpleAuth();
-        }
+                MQTT mqtt = buildClient(ep);
+                connection = mqtt.futureConnection();
+                connections.add(connection);
                log.info("mqtt endpoint [{}] connect sending", ep.getName());
-        CompletableFuture<Mqtt3ConnAck> timeoutProbe = connectBuilder.send()
-                .orTimeout(Math.max(15, ep.getConnectionTimeoutSeconds() + 5L), TimeUnit.SECONDS);
-        timeoutProbe.whenComplete((ack, err) -> {
-            if (err != null) {
-                log.error("mqtt endpoint [{}] connect failed", ep.getName(), err);
-                publishOperationalError(ep, "connect", err.getMessage() == null ? err.getClass().getSimpleName() : err.getMessage());
+                connection.connect().await();
+                log.info("mqtt endpoint [{}] connected", ep.getName());
+                connection.subscribe(new Topic[]{new Topic(ep.getTopic(), mapQos(ep.getQos()))}).await();
+                log.info("mqtt endpoint [{}] subscribed topic={} qos={}", ep.getName(), ep.getTopic(), ep.getQos());
+                receiveLoop(ep, connection);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                return;
-            }
-            log.info("mqtt endpoint [{}] connected code={}", ep.getName(), ack.getReturnCode());
-            client.subscribeWith()
-                    .topicFilter(ep.getTopic())
-                    .qos(mapQos(ep.getQos()))
-                    .callback(publish -> onMessage(ep, publish.getTopic().toString(), publish.getPayloadAsBytes()))
-                    .send()
-                    .whenComplete((subAck, subErr) -> {
-                        if (subErr != null) {
-                            log.error("mqtt endpoint [{}] subscribe failed topic={}", ep.getName(), ep.getTopic(), subErr);
-                            publishOperationalError(ep, "subscribe",
-                                    subErr.getMessage() == null ? subErr.getClass().getSimpleName() : subErr.getMessage());
+            } catch (Exception e) {
+                if (isPeerDisconnect(e)) {
+                    log.warn("mqtt endpoint [{}] peer disconnected, reconnecting: {}", ep.getName(), e.getMessage());
                } else {
-                            log.info("mqtt endpoint [{}] subscribed topic={} qos={}",
-                                    ep.getName(), ep.getTopic(), ep.getQos());
+                    log.error("mqtt endpoint [{}] connect/receive failed", ep.getName(), e);
+                    publishOperationalError(ep, "connect",
+                            e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
+                }
+                sleepBeforeReconnect(ep);
+            } finally {
+                if (connection != null) {
+                    connections.remove(connection);
+                    try {
+                        connection.disconnect().await();
+                    } catch (Exception ignored) {
+                        // best-effort close
+                    }
+                }
+            }
+        }
+    }
+
+    private MQTT buildClient(MqttInboundProperties.Endpoint ep) throws Exception {
+        MQTT mqtt = new MQTT();
+        mqtt.setHost(ep.getUri());
+        if (ep.getClientId() != null && !ep.getClientId().isBlank()) {
+            mqtt.setClientId(ep.getClientId());
+        }
+        mqtt.setCleanSession(ep.isCleanSession());
+        if (ep.getUsername() != null && !ep.getUsername().isBlank()) {
+            mqtt.setUserName(ep.getUsername());
+            mqtt.setPassword(ep.getPassword() == null ? "" : ep.getPassword());
+        }
+        mqtt.setConnectAttemptsMax(1);
+        mqtt.setReconnectAttemptsMax(0);
+        if (ep.getKeepAliveSeconds() > 0) {
+            mqtt.setKeepAlive((short) ep.getKeepAliveSeconds());
+        }
+        mqtt.setVersion("3.1.1");
+        if (hasCustomTls(ep.getTls())) {
+            mqtt.setSslContext(MqttTlsSupport.buildSslContext(ep.getTls()));
+        }
+        return mqtt;
+    }
+
+    private void receiveLoop(MqttInboundProperties.Endpoint ep, FutureConnection connection) throws Exception {
+        while (running && !Thread.currentThread().isInterrupted()) {
+            Future<Message> futureMessage = connection.receive();
+            Message message = futureMessage.await();
+            try {
+                message.ack();
+                onMessage(ep, message.getTopic(), message.getPayload());
+            } catch (Exception e) {
+                log.warn("mqtt endpoint [{}] message ack/dispatch failed topic={} len={}",
+                        ep.getName(), message.getTopic(), message.getPayload() == null ? 0 : message.getPayload().length, e);
+            }
+        }
+    }
+
+    private boolean isPeerDisconnect(Exception e) {
+        Throwable current = e;
+        while (current != null) {
+            if (current instanceof EOFException) {
+                return true;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
+
+    private void sleepBeforeReconnect(MqttInboundProperties.Endpoint ep) {
+        try {
+            Thread.sleep(5000);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
        }
-                    });
-        });
    }

    private void onMessage(MqttInboundProperties.Endpoint ep, String topic, byte[] payload) {
@@ -298,19 +319,11 @@ public final class MqttEndpointManager implements AutoCloseable {
        return value.replace("\\", "\\\\").replace("\"", "\\\"");
    }

-    private static MqttQos mapQos(int level) {
+    private static QoS mapQos(int level) {
        return switch (level) {
-            case 0 -> MqttQos.AT_MOST_ONCE;
-            case 2 -> MqttQos.EXACTLY_ONCE;
-            default -> MqttQos.AT_LEAST_ONCE;
-        };
-    }
-
-    private static int defaultPort(String scheme) {
-        if (scheme == null) return 1883;
-        return switch (scheme.toLowerCase()) {
-            case "ssl", "tls", "mqtts" -> 8883;
-            default -> 1883;
+            case 0 -> QoS.AT_MOST_ONCE;
+            case 2 -> QoS.EXACTLY_ONCE;
+            default -> QoS.AT_LEAST_ONCE;
        };
    }

@@ -318,6 +331,13 @@ public final class MqttEndpointManager implements AutoCloseable {
        return value != null && value.trim().length() == 17;
    }

+    private static String safeName(String value) {
+        if (value == null || value.isBlank()) {
+            return "unnamed";
+        }
+        return value.replaceAll("[^a-zA-Z0-9_.-]", "_");
+    }
+
    private record IdentityResolution(VehicleIdentity identity, String errorMessage) {}

    private static boolean hasCustomTls(MqttInboundProperties.Tls tls) {
@@ -331,13 +351,17 @@ public final class MqttEndpointManager implements AutoCloseable {

    @Override
    public void close() {
-        for (Mqtt3AsyncClient c : clients) {
+        running = false;
+        for (FutureConnection connection : List.copyOf(connections)) {
            try {
-                c.disconnect();
+                connection.disconnect().await();
            } catch (Exception ignored) {
                // best-effort
            }
        }
-        log.info("mqtt endpoint manager stopped, clients={}", clients.size());
+        for (Thread worker : workers) {
+            worker.interrupt();
+        }
+        log.info("mqtt endpoint manager stopped, connections={}", connections.size());
    }
 }
--- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java
+++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java
@@ -1,18 +1,18 @@
 package com.lingniu.ingest.inbound.mqtt.client;

-import com.hivemq.client.mqtt.MqttClientSslConfig;
-import com.hivemq.client.mqtt.MqttClientSslConfigBuilder;
 import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;

 import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManagerFactory;
-import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.KeyFactory;
 import java.security.KeyStore;
 import java.security.PrivateKey;
+import java.security.SecureRandom;
 import java.security.cert.Certificate;
 import java.security.cert.CertificateFactory;
 import java.security.spec.PKCS8EncodedKeySpec;
@@ -25,74 +25,93 @@ public final class MqttTlsSupport {

    private MqttTlsSupport() {}

-    public static MqttClientSslConfig build(MqttInboundProperties.Tls tls) {
+    public static SSLContext buildSslContext(MqttInboundProperties.Tls tls) {
        try {
-            MqttClientSslConfigBuilder builder = MqttClientSslConfig.builder();
+            TrustManagerFactory tmf = null;
+            KeyManagerFactory kmf = null;
            if (tls != null && hasText(tls.getCaPem())) {
-                builder.trustManagerFactory(trustManagerFactory(Path.of(tls.getCaPem())));
+                tmf = trustManagerFactory(Path.of(tls.getCaPem()));
            }
            if (tls != null && hasText(tls.getClientPem()) && hasText(tls.getClientKey())) {
-                builder.keyManagerFactory(keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey())));
+                kmf = keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey()));
            }
-            if (tls != null && !tls.isHostnameVerificationEnabled()) {
-                builder.hostnameVerifier((hostname, session) -> true);
-            }
-            return builder.build();
+            SSLContext context = SSLContext.getInstance("TLSv1.2");
+            context.init(
+                    kmf == null ? null : kmf.getKeyManagers(),
+                    tmf == null ? null : tmf.getTrustManagers(),
+                    new SecureRandom());
+            return context;
        } catch (Exception e) {
            throw new IllegalStateException("mqtt tls config build failed", e);
        }
    }

-    private static TrustManagerFactory trustManagerFactory(Path caPem) throws Exception {
+    private static TrustManagerFactory trustManagerFactory(Path caPath) throws Exception {
        KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
        trustStore.load(null, null);
        int i = 0;
-        for (Certificate cert : readCertificates(caPem)) {
+        for (Certificate cert : readCertificates(caPath)) {
            trustStore.setCertificateEntry("ca-" + i++, cert);
        }
-        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+        TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
        tmf.init(trustStore);
        return tmf;
    }

-    private static KeyManagerFactory keyManagerFactory(Path clientPem, Path clientKey) throws Exception {
-        List<Certificate> chain = readCertificates(clientPem);
-        PrivateKey key = readPrivateKey(clientKey);
+    private static KeyManagerFactory keyManagerFactory(Path clientCertPath, Path clientKeyPath) throws Exception {
+        List<Certificate> chain = readCertificates(clientCertPath);
+        PrivateKey key = readPrivateKey(clientKeyPath);
        KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
        keyStore.load(null, null);
        char[] password = new char[0];
-        keyStore.setKeyEntry("mqtt-client", key, password, chain.toArray(Certificate[]::new));
-        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyStore.setCertificateEntry("client", chain.getFirst());
+        keyStore.setKeyEntry("clientKey", key, password, chain.toArray(Certificate[]::new));
+        KeyManagerFactory kmf = KeyManagerFactory.getInstance("PKIX");
        kmf.init(keyStore, password);
        return kmf;
    }

-    private static List<Certificate> readCertificates(Path pem) throws Exception {
-        String text = Files.readString(pem, StandardCharsets.UTF_8);
-        List<Certificate> certificates = new ArrayList<>();
+    private static List<Certificate> readCertificates(Path path) throws Exception {
        CertificateFactory factory = CertificateFactory.getInstance("X.509");
+        String text = readUtf8IfPossible(path);
+        if (text != null && text.contains("-----BEGIN CERTIFICATE-----")) {
+            List<Certificate> certificates = new ArrayList<>();
            for (String block : pemBlocks(text, "CERTIFICATE")) {
-            try (ByteArrayInputStream in = new ByteArrayInputStream(Base64.getMimeDecoder().decode(block))) {
-                certificates.add(factory.generateCertificate(in));
-            }
-        }
-        if (certificates.isEmpty()) {
-            throw new IllegalArgumentException("no certificate found in " + pem);
+                certificates.add(factory.generateCertificate(
+                        new java.io.ByteArrayInputStream(Base64.getMimeDecoder().decode(block))));
            }
+            if (!certificates.isEmpty()) {
                return certificates;
            }
+        }
+        try (InputStream in = Files.newInputStream(path)) {
+            Collection<? extends Certificate> certificates = factory.generateCertificates(in);
+            if (certificates.isEmpty()) {
+                throw new IllegalArgumentException("no certificate found in " + path);
+            }
+            return List.copyOf(certificates);
+        }
+    }

-    private static PrivateKey readPrivateKey(Path pem) throws Exception {
-        String text = Files.readString(pem, StandardCharsets.UTF_8);
+    private static PrivateKey readPrivateKey(Path path) throws Exception {
+        String text = Files.readString(path, StandardCharsets.UTF_8);
        List<String> blocks = pemBlocks(text, "PRIVATE KEY");
        if (blocks.isEmpty()) {
-            throw new IllegalArgumentException("no PKCS#8 private key found in " + pem);
+            throw new IllegalArgumentException("no PKCS#8 private key found in " + path);
        }
        byte[] der = Base64.getMimeDecoder().decode(blocks.getFirst());
        PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(der);
        return KeyFactory.getInstance("RSA").generatePrivate(spec);
    }

+    private static String readUtf8IfPossible(Path path) {
+        try {
+            return Files.readString(path, StandardCharsets.UTF_8);
+        } catch (Exception ignored) {
+            return null;
+        }
+    }
+
    private static List<String> pemBlocks(String text, String type) {
        String begin = "-----BEGIN " + type + "-----";
        String end = "-----END " + type + "-----";
--- 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
@@ -172,7 +172,7 @@ class MqttEndpointManagerTest {
                    .containsEntry("endpoint", "bad-endpoint")
                    .containsEntry("profile", "yutong")
                    .containsEntry("operationalError", "true")
-                    .containsEntry("phase", "init")
+                    .containsEntry("phase", "connect")
                    .containsEntry("identityResolved", "false")
                    .containsEntry("identitySource", "UNKNOWN");
        });
@@ -184,7 +184,7 @@ class MqttEndpointManagerTest {
                    .containsEntry("operationalError", "true")
                    .containsEntry("endpoint", "bad-endpoint")
                    .containsEntry("profile", "yutong")
-                    .containsEntry("phase", "init");
+                    .containsEntry("phase", "connect");
        });

        batchExecutor.close();
--- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java
+++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java
@@ -1,6 +1,5 @@
 package com.lingniu.ingest.inbound.mqtt;

-import com.hivemq.client.mqtt.MqttClientSslConfig;
 import com.lingniu.ingest.inbound.mqtt.client.MqttTlsSupport;
 import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
 import org.junit.jupiter.api.Test;
@@ -9,6 +8,8 @@ import org.junit.jupiter.api.io.TempDir;
 import java.nio.file.Files;
 import java.nio.file.Path;

+import javax.net.ssl.SSLContext;
+
 import static org.assertj.core.api.Assertions.assertThat;

 class MqttTlsSupportTest {
@@ -23,10 +24,9 @@ class MqttTlsSupportTest {
        MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls();
        tls.setCaPem(ca.toString());

-        MqttClientSslConfig config = MqttTlsSupport.build(tls);
+        SSLContext config = MqttTlsSupport.buildSslContext(tls);

-        assertThat(config.getTrustManagerFactory()).isPresent();
-        assertThat(config.getKeyManagerFactory()).isEmpty();
+        assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
    }

    @Test
@@ -42,24 +42,22 @@ class MqttTlsSupportTest {
        tls.setClientPem(cert.toString());
        tls.setClientKey(key.toString());

-        MqttClientSslConfig config = MqttTlsSupport.build(tls);
+        SSLContext config = MqttTlsSupport.buildSslContext(tls);

-        assertThat(config.getTrustManagerFactory()).isPresent();
-        assertThat(config.getKeyManagerFactory()).isPresent();
-        assertThat(config.getHostnameVerifier()).isEmpty();
+        assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
    }

    @Test
-    void canDisableHostnameVerificationForLegacyMutualTlsBrokers() throws Exception {
+    void hostnameVerificationFlagDoesNotBlockSslContextCreation() throws Exception {
        Path ca = tempDir.resolve("ca.pem");
        Files.writeString(ca, CERT);
        MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls();
        tls.setCaPem(ca.toString());
        tls.setHostnameVerificationEnabled(false);

-        MqttClientSslConfig config = MqttTlsSupport.build(tls);
+        SSLContext config = MqttTlsSupport.buildSslContext(tls);

-        assertThat(config.getHostnameVerifier()).isPresent();
+        assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
    }

    private static final String CERT = """
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
@@ -27,6 +27,7 @@ import java.util.List;
 */
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(Gb32960DecodedFrameService.class)
@RequestMapping("/api/event-history/gb32960")
@Tag(name = "gb-32960-frame-controller", description = "GB32960 专用历史帧、业务快照、字段查询和字段字典接口。")
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java
@@ -26,6 +26,7 @@ import java.util.List;

@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(TdengineHistoryReader.class)
@RequestMapping("/api/event-history/jt808")
@Tag(name = "jt-808-location-controller", description = "JT808 位置历史分页查询接口。")
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java
@@ -25,6 +25,7 @@ import java.util.Locale;

@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(TdengineHistoryReader.class)
@RequestMapping("/api/event-history/jt808")
@Tag(name = "jt-808-raw-frame-controller", description = "JT808 原始帧索引分页查询接口。")
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/LocationHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/LocationHistoryController.java
@@ -0,0 +1,204 @@
+package com.lingniu.ingest.eventhistory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
+import com.lingniu.ingest.tdenginehistory.TdenginePage;
+import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
+import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+import java.util.Locale;
+
+@RestController
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnBean(TdengineHistoryReader.class)
+@RequestMapping("/api/event-history")
+@Tag(name = "location-history-controller", description = "通用位置历史分页查询接口。")
+public final class LocationHistoryController {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private final TdengineHistoryReader reader;
+
+    public LocationHistoryController(TdengineHistoryReader reader) {
+        if (reader == null) {
+            throw new IllegalArgumentException("reader must not be null");
+        }
+        this.reader = reader;
+    }
+
+    @GetMapping("/locations")
+    @Operation(
+            summary = "查询车辆位置历史",
+            description = "按协议、车辆标识和时间范围查询 TDengine 位置点。分页使用 cursorTs + cursorId，不使用 offset。")
+    public LocationPageResponse locations(
+            @Parameter(description = "协议名，例如 GB32960、JT808、MQTT_YUTONG、XINDA_PUSH。", required = true, example = "MQTT_YUTONG")
+            @RequestParam String protocol,
+            @Parameter(description = "内部车辆键。传入后优先使用。", example = "LMRKH9AC2R1004087")
+            @RequestParam(required = false) String vehicleKey,
+            @Parameter(description = "VIN。GB32960、宇通、信达通常可直接用 VIN。", example = "LMRKH9AC2R1004087")
+            @RequestParam(required = false) String vin,
+            @Parameter(description = "JT808 终端手机号或终端标识；JT808 会自动映射为 jt808:<phone>。", example = "g7gps")
+            @RequestParam(required = false) String phone,
+            @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00")
+            @RequestParam String dateFrom,
+            @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00")
+            @RequestParam String dateTo,
+            @Parameter(description = "排序方向。", example = "DESC")
+            @RequestParam(defaultValue = "DESC") TdengineQueryOrder order,
+            @Parameter(description = "返回位置点数量上限，最大 1000。", example = "100")
+            @RequestParam(defaultValue = "100") int limit,
+            @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z")
+            @RequestParam(required = false) String cursorTs,
+            @Parameter(description = "上一页返回的 nextCursor.id。", example = "location-xxx")
+            @RequestParam(required = false) String cursorId) throws IOException {
+        String normalizedProtocol = require(protocol, "protocol is required for location query").toUpperCase(Locale.ROOT);
+        QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
+        TdenginePage<TdengineLocationRow> page = reader.queryLocations(new TdengineLocationQuery(
+                normalizedProtocol,
+                resolveVehicleKey(normalizedProtocol, vehicleKey, vin, phone),
+                range.eventTimeFrom(),
+                range.eventTimeTo().plusMillis(1),
+                order,
+                limit,
+                cursor(cursorTs, cursorId)));
+        return LocationPageResponse.from(page);
+    }
+
+    private static String resolveVehicleKey(String protocol, String vehicleKey, String vin, String phone) {
+        String explicitVehicleKey = trimToNull(vehicleKey);
+        if (explicitVehicleKey != null) {
+            return explicitVehicleKey;
+        }
+        String vinValue = trimToNull(vin);
+        if (vinValue != null) {
+            return vinValue;
+        }
+        String phoneValue = trimToNull(phone);
+        if (phoneValue != null) {
+            if ("JT808".equals(protocol) && !phoneValue.startsWith("jt808:")) {
+                return "jt808:" + phoneValue;
+            }
+            return phoneValue;
+        }
+        throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vehicleKey, vin or phone is required");
+    }
+
+    private static String require(String value, String message) {
+        String trimmed = trimToNull(value);
+        if (trimmed == null) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
+        }
+        return trimmed;
+    }
+
+    private static String trimToNull(String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        return value.trim();
+    }
+
+    private static TdenginePageCursor cursor(String cursorTs, String cursorId) {
+        boolean hasTs = cursorTs != null && !cursorTs.isBlank();
+        boolean hasId = cursorId != null && !cursorId.isBlank();
+        if (!hasTs && !hasId) {
+            return null;
+        }
+        if (!hasTs || !hasId) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together");
+        }
+        return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim());
+    }
+
+    public record LocationPageResponse(
+            List<LocationResponse> items,
+            CursorResponse nextCursor) {
+
+        private static LocationPageResponse from(TdenginePage<TdengineLocationRow> page) {
+            CursorResponse next = page.nextCursor()
+                    .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
+                    .orElse(null);
+            return new LocationPageResponse(
+                    page.items().stream().map(LocationResponse::from).toList(),
+                    next);
+        }
+    }
+
+    public record CursorResponse(String ts, String id) {
+    }
+
+    public record LocationResponse(
+            String eventTime,
+            String receivedAt,
+            String factId,
+            String frameId,
+            String phone,
+            String vin,
+            String vehicleKey,
+            double longitude,
+            double latitude,
+            double altitudeM,
+            double speedKmh,
+            double directionDeg,
+            long alarmFlag,
+            long statusFlag,
+            Double totalMileageKm,
+            String rawUri,
+            String metadataJson,
+            String protocol) {
+
+        private static LocationResponse from(TdengineLocationRow row) {
+            return new LocationResponse(
+                    row.ts().toString(),
+                    row.receivedAt().toString(),
+                    row.factId(),
+                    row.frameId(),
+                    row.phone(),
+                    row.vin(),
+                    row.vehicleKey(),
+                    row.longitude(),
+                    row.latitude(),
+                    row.altitudeM(),
+                    row.speedKmh(),
+                    row.directionDeg(),
+                    row.alarmFlag(),
+                    row.statusFlag(),
+                    row.totalMileageKm(),
+                    rawUri(row),
+                    row.metadataJson(),
+                    row.protocol());
+        }
+
+        private static String rawUri(TdengineLocationRow row) {
+            if (row.rawUri() != null && !row.rawUri().isBlank()) {
+                return row.rawUri();
+            }
+            String metadataJson = row.metadataJson();
+            if (metadataJson == null || metadataJson.isBlank()) {
+                return row.rawUri();
+            }
+            try {
+                return OBJECT_MAPPER.readTree(metadataJson).path("rawArchiveUri").asText(row.rawUri());
+            } catch (JsonProcessingException ignored) {
+                return row.rawUri();
+            }
+        }
+    }
+}
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
@@ -9,6 +9,7 @@ import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
 import com.lingniu.ingest.eventhistory.Gb32960FrameController;
 import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
 import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
+import com.lingniu.ingest.eventhistory.LocationHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
 import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
@@ -95,6 +96,7 @@ public class EventHistoryAutoConfiguration {

    @Bean
    @ConditionalOnBean(Gb32960DecodedFrameService.class)
+    @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
    @ConditionalOnMissingBean
    public Gb32960FrameController gb32960FrameController(Gb32960DecodedFrameService service) {
        return new Gb32960FrameController(service);
@@ -103,12 +105,21 @@ public class EventHistoryAutoConfiguration {
    @Bean
    @ConditionalOnBean(TdengineHistoryReader.class)
    @ConditionalOnMissingBean
+    public LocationHistoryController locationHistoryController(TdengineHistoryReader reader) {
+        return new LocationHistoryController(reader);
+    }
+
+    @Bean
+    @ConditionalOnBean(TdengineHistoryReader.class)
+    @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
+    @ConditionalOnMissingBean
    public Jt808LocationHistoryController jt808LocationHistoryController(TdengineHistoryReader reader) {
        return new Jt808LocationHistoryController(reader);
    }

    @Bean
    @ConditionalOnBean(TdengineHistoryReader.class)
+    @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
    @ConditionalOnMissingBean
    public Jt808RawFrameHistoryController jt808RawFrameHistoryController(TdengineHistoryReader reader) {
        return new Jt808RawFrameHistoryController(reader);
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java
@@ -0,0 +1,113 @@
+package com.lingniu.ingest.eventhistory;
+
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
+import com.lingniu.ingest.tdenginehistory.TdenginePage;
+import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
+import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LocationHistoryControllerTest {
+
+    @Test
+    void queriesGenericLocationHistoryWithCursor() throws Exception {
+        AtomicReference<TdengineLocationQuery> captured = new AtomicReference<>();
+        TdengineHistoryReader reader = new StubReader(captured, new TdenginePage<>(
+                List.of(row("fact-1")),
+                Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "fact-1"))));
+        LocationHistoryController controller = new LocationHistoryController(reader);
+
+        LocationHistoryController.LocationPageResponse response = controller.locations(
+                "mqtt_yutong",
+                null,
+                "LMRKH9AC2R1004087",
+                null,
+                "2026-06-29T13:00:00",
+                "2026-06-29T13:10:00",
+                TdengineQueryOrder.DESC,
+                100,
+                null,
+                null);
+
+        assertThat(captured.get().protocol()).isEqualTo("MQTT_YUTONG");
+        assertThat(captured.get().vehicleKey()).isEqualTo("LMRKH9AC2R1004087");
+        assertThat(response.items()).extracting(LocationHistoryController.LocationResponse::factId)
+                .containsExactly("fact-1");
+        assertThat(response.nextCursor()).isEqualTo(new LocationHistoryController.CursorResponse(
+                "2026-06-29T05:00:01Z", "fact-1"));
+    }
+
+    @Test
+    void mapsJt808PhoneToVehicleKey() throws Exception {
+        AtomicReference<TdengineLocationQuery> captured = new AtomicReference<>();
+        LocationHistoryController controller = new LocationHistoryController(
+                new StubReader(captured, new TdenginePage<>(List.of(), null)));
+
+        controller.locations(
+                "JT808",
+                null,
+                null,
+                "g7gps",
+                "2026-06-29T13:00:00",
+                "2026-06-29T13:10:00",
+                TdengineQueryOrder.DESC,
+                100,
+                null,
+                null);
+
+        assertThat(captured.get().vehicleKey()).isEqualTo("jt808:g7gps");
+    }
+
+    private static TdengineLocationRow row(String factId) {
+        Instant ts = Instant.parse("2026-06-29T05:00:00Z");
+        return new TdengineLocationRow(
+                ts,
+                factId,
+                "frame-1",
+                ts,
+                121.1,
+                30.5,
+                0.0,
+                32.0,
+                90.0,
+                0,
+                1,
+                null,
+                "",
+                "{\"rawArchiveUri\":\"archive://raw.bin\"}",
+                "MQTT_YUTONG",
+                "LMRKH9AC2R1004087",
+                "LMRKH9AC2R1004087",
+                "");
+    }
+
+    private record StubReader(AtomicReference<TdengineLocationQuery> captured,
+                              TdenginePage<TdengineLocationRow> page) implements TdengineHistoryReader {
+
+        @Override
+        public com.lingniu.ingest.tdenginehistory.TdenginePage<com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow> queryRawFrames(
+                com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery query) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) {
+            captured.set(query);
+            return page;
+        }
+
+        @Override
+        public com.lingniu.ingest.tdenginehistory.TdenginePage<com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow> queryTelemetryFields(
+                com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery query) {
+            throw new UnsupportedOperationException();
+        }
+    }
+}
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
@@ -9,6 +9,7 @@ import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
 import com.lingniu.ingest.eventhistory.Gb32960FrameController;
 import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
 import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
+import com.lingniu.ingest.eventhistory.LocationHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
 import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
 import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
@@ -58,6 +59,7 @@ class EventHistoryAutoConfigurationTest {
        contextRunner
                .withPropertyValues(
                        "lingniu.ingest.event-history.enabled=true",
+                        "lingniu.ingest.event-history.api.specialized-enabled=true",
                        "lingniu.ingest.sink.archive.path=/tmp/lingniu-test-archive")
                .withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
                .withBean(SinkArchiveProperties.class, SinkArchiveProperties::new)
@@ -73,6 +75,7 @@ class EventHistoryAutoConfigurationTest {
                .withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
                .withPropertyValues(
                        "lingniu.ingest.event-history.enabled=true",
+                        "lingniu.ingest.event-history.api.specialized-enabled=true",
                        "lingniu.ingest.sink.archive.path=/tmp/lingniu-test-archive")
                .withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
@@ -87,16 +90,33 @@ class EventHistoryAutoConfigurationTest {
    @Test
    void createsJt808LocationHistoryControllerWhenTdengineReaderExists() {
        contextRunner
-                .withPropertyValues("lingniu.ingest.event-history.enabled=true")
+                .withPropertyValues(
+                        "lingniu.ingest.event-history.enabled=true",
+                        "lingniu.ingest.event-history.api.specialized-enabled=true")
                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
                .run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class));
    }

    @Test
-    void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() {
+    void createsGenericLocationAndTelemetryControllersWhenTdengineReaderExists() {
        contextRunner
                .withPropertyValues("lingniu.ingest.event-history.enabled=true")
                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
+                .run(context -> {
+                    assertThat(context).hasSingleBean(LocationHistoryController.class);
+                    assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class);
+                    assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
+                    assertThat(context).doesNotHaveBean(Jt808RawFrameHistoryController.class);
+                });
+    }
+
+    @Test
+    void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() {
+        contextRunner
+                .withPropertyValues(
+                        "lingniu.ingest.event-history.enabled=true",
+                        "lingniu.ingest.event-history.api.specialized-enabled=true")
+                .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
                .run(context -> assertThat(context).hasSingleBean(Jt808RawFrameHistoryController.class));
    }

--- a/pom.xml
+++ b/pom.xml
@@ -65,7 +65,7 @@
        <resilience4j.version>2.2.0</resilience4j.version>
        <micrometer.version>1.13.6</micrometer.version>
        <otel.version>1.42.1</otel.version>
-        <hivemq-mqtt.version>1.3.3</hivemq-mqtt.version>
+        <fusesource-mqtt.version>1.12</fusesource-mqtt.version>
        <duckdb.version>1.1.3</duckdb.version>
        <taos-jdbcdriver.version>3.8.4</taos-jdbcdriver.version>
        <springdoc-openapi.version>2.8.17</springdoc-openapi.version>
@@ -307,9 +307,9 @@
                <version>${kafka.version}</version>
            </dependency>
            <dependency>
-                <groupId>com.hivemq</groupId>
-                <artifactId>hivemq-mqtt-client</artifactId>
-                <version>${hivemq-mqtt.version}</version>
+                <groupId>org.fusesource.mqtt-client</groupId>
+                <artifactId>mqtt-client</artifactId>
+                <version>${fusesource-mqtt.version}</version>
            </dependency>
            <dependency>
                <groupId>org.duckdb</groupId>