fix: resolve merge conflict in MqttGenericConsumer

Remote branch overwrote the AbstractMqttConsumer inheritance structure.
Restored extends AbstractMqttConsumer + constructor injection while
preserving remote's business logic changes (updatecontrollerAccept,
SimpleDateFormat conversion, 6-arg insertsyscontroller).

iot-platform/src/main/java/com/iot/platform/mqtt/MqttGenericConsumer.java

@@ -1,50 +1,25 @@
 package com.iot.platform.mqtt;
-import com.alibaba.fastjson2.util.DateUtils;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.iot.platform.domain.ControllerData;
 import com.iot.platform.domain.topics;
 import com.iot.platform.service.SysControllerService;
-import com.iot.platform.service.TDengineService;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import com.iot.platform.config.IotProperties;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.net.ssl.*;
-import java.io.InputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.security.KeyStore;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
-import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
+import java.text.SimpleDateFormat;
+import java.util.Date;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * 存储控制器数据
  */
 @Component
-public class MqttGenericConsumer {
+public class MqttGenericConsumer extends AbstractMqttConsumer {
 
-    private static final Logger log = LoggerFactory.getLogger(MqttGenericConsumer.class);
-    private static ExecutorService threadPool= Executors.newCachedThreadPool();
-    /**
-     *
-     * 添加mysql数据
-     *
-     */
     @Autowired
     private StringRedisTemplate stringRedisTemplate;
     @Autowired
@@ -52,8 +27,7 @@ public class MqttGenericConsumer {
     @Autowired
     public MqttDynamicConsumer messageListenerService2;
 
-    @Autowired
-    private IotProperties iotProperties;
+    private final ObjectMapper objectMapper = new ObjectMapper();
 
     @Autowired
     public MqttGenericConsumer(@Qualifier("abstractConsumerExecutor") ExecutorService executorService) {
@@ -65,141 +39,16 @@ public class MqttGenericConsumer {
         return "+/generics";
     }
 
-    /**
-     * 检测普通MQTT服务器连通性（移除SSL，仅验证TCP端口可达性）
-     */
-    private void checkServerAvailability() throws InterruptedException {
-        boolean serverAvailable = false;
-        while (!serverAvailable) {
-            try (Socket socket = new Socket()) {
-                // 改用普通brokerHost和brokerPort
-                socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-                serverAvailable = true;
-                log.info("普通MQTT服务器连通性检测通过");
-            } catch (Exception e) {
-                log.error("普通MQTT服务器不可达，5秒后重试...");
-                Thread.sleep(RECONNECT_INTERVAL);
-            }
-        }
-    }
-
-    /**
-     * 初始化MQTT连接选项（移除SSL配置，新增账号密码认证）
-     */
-    private void initMqttConnectOptions() {
-        connOpts = new MqttConnectOptions();
-        connOpts.setCleanSession(true);
-        connOpts.setAutomaticReconnect(true);
-        connOpts.setConnectionTimeout(10);
-
-        // ========== 新增：配置MQTT账号密码 ==========
-        connOpts.setUserName(mqttUsername);
-        connOpts.setPassword(mqttPassword.toCharArray()); // 密码要求传入char数组
-
-        // 移除：原有的SSL配置方法调用（configureSslAndStrictHostnameVerify()）
-    }
-
-    /**
-     * 按操作系统生成唯一ClientId（保持原有逻辑不变）
-     */
-    private String generateClientIdByOs() {
+    @Override
+    protected String generateClientId() {
         String osName = System.getProperty("os.name").toLowerCase();
         return osName.contains("windows") ? "mqttx_e216fbf1615" : "mqttx_e216fbf1616";
     }
 
-    /**
-     * 设置MQTT回调函数（保留核心逻辑，修复2个关键bug）
-     */
-    private void setMqttCallback() {
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT连接断开，开始重连：" + cause.getMessage());
-                isMqttConnected = false;
-                reconnect();
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) throws Exception {
-                if (isMqttConnected) {
-                    executorService.submit(() -> {
-                        try {
-                            ObjectMapper objectMapper = new ObjectMapper();
-                            // 修复bug1：原代码先读取正确的messageContent，却误用mqtt（message.toString()）解析
-                            // 正确读取消息负载（UTF-8编码，避免乱码）
-                            String messageContent = new String(message.getPayload(), "UTF-8");
-                            // 修复bug2：用正确的messageContent解析为ControllerData对象
-                            ControllerData controllerData = objectMapper.readValue(messageContent, ControllerData.class);
-                            // 业务处理
-                            triggermethod(controllerData);
-                        } catch (Exception e) {
-                            log.error("消息处理失败：", e);
-                        }
-                    });
-                }
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                // 消息投递完成回调（无需处理）
-            }
-        });
-    }
-
-    /**
-     * 建立MQTT连接并订阅主题（保持逻辑不变，移除SSL日志标识）
-     */
-    private void connectAndSubscribeTopic() throws MqttException {
-        if (!mqttClient.isConnected()) {
-            IMqttToken connectToken = mqttClient.connectWithResult(connOpts);
-            if (connectToken.isComplete()) {
-                mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                isMqttConnected = true;
-                log.info("MQTT连接成功，已订阅主题：" + SUBSCRIBE_TOPIC);
-            }
-        }
-    }
-
-    /**
-     * MQTT重连逻辑（移除SSL日志标识，保持功能不变）
-     */
-    public void reconnect() {
-        int maxReconnectAttempts = 3;
-        for (int attempt = 1; attempt <= maxReconnectAttempts; attempt++) {
-            try {
-                Thread.sleep(RECONNECT_INTERVAL);
-                if (mqttClient != null && !mqttClient.isConnected()) {
-                    mqttClient.connect(connOpts);
-                    mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                    isMqttConnected = true;
-                    log.info("MQTT重连成功（第" + attempt + "次尝试）");
-                    break; // 重连成功后退出循环
-                }
-            } catch (MqttException | InterruptedException e) {
-                log.error("MQTT重连失败（第" + attempt + "次尝试）：" + e.getMessage());
-                if (attempt == maxReconnectAttempts) {
-                    log.error("已达最大重连次数，停止重连");
-                }
-            }
-        }
-    }
-
-    /**
-     * 销毁时断开MQTT连接，关闭线程池（移除未定义的threadPool.shutdown()）
-     */
-    @PreDestroy
-    public void disconnect() {
-        try {
-            if (mqttClient != null && mqttClient.isConnected()) {
-                mqttClient.disconnect();
-                mqttClient.close();
-                log.info("MQTT连接已断开");
-            }
-            executorService.shutdown();
-            // 移除：原代码中未定义的threadPool.shutdown();（避免编译错误）
-        } catch (MqttException e) {
-            log.error("MQTT断开连接失败：", e);
-        }
+    @Override
+    protected void handleMessage(String topic, String messageContent) throws Exception {
+        ControllerData controllerData = objectMapper.readValue(messageContent, ControllerData.class);
+        triggermethod(controllerData);
     }
 
     public void triggermethod(ControllerData weather) throws Exception {
@@ -209,48 +58,42 @@ public class MqttGenericConsumer {
         List<topics> topics = weather.getTopics();
         List<topics> cmdtopics = weather.getCmd_topics();
         topics faultprot = weather.getFault_prot();
-        //需要检索全部的数据是否存在，如果存在就进行修改，如果不存在就进行添加
-        Integer controllercountcount=0;
+
+        Integer controllercountcount = 0;
         for (topics topicsMap : topics) {
             Integer count = sysControllerService.selectsyscontrollercount(topicsMap.getPath());
             if (count <= 0) {
-                //存储redis
                 stringRedisTemplate.persist(controllerId);
-                stringRedisTemplate.opsForHash().put(controllerId+":"+topicsMap.getName(), "path", topicsMap.getPath());
-                //将数据存储到mysql中
-                sysControllerService.insertsyscontroller(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath(),topicsMap.getPath().split("/")[1]);
+                stringRedisTemplate.opsForHash().put(controllerId + ":" + topicsMap.getName(), "path", topicsMap.getPath());
+                sysControllerService.insertsyscontroller(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath(), topicsMap.getPath().split("/")[1]);
                 controllercountcount++;
-            }else{
-                // 毫秒时间戳转换为秒级日期格式
+            } else {
                 long ts = Long.parseLong(timestamp);
-                String date = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
-                        .format(new java.util.Date(ts));
-                //修改数据库
-                sysControllerService.updatecontrollerAccept(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath(),topicsMap.getPath().split("/")[1],date);
+                String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(ts));
+                sysControllerService.updatecontrollerAccept(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath(), topicsMap.getPath().split("/")[1], date);
             }
         }
-        Integer controllercountcmdcount=0;
+
+        Integer controllercountcmdcount = 0;
         for (topics cmdtopicsMap : cmdtopics) {
-            //将数据存储到redis
             Integer count = sysControllerService.selectsyscontrollercountcmd(cmdtopicsMap.getPath());
-            if (count<=0) {
-                stringRedisTemplate.opsForHash().put(controllerId+"_cmd:"+cmdtopicsMap.getName(), "path", cmdtopicsMap.getPath());
+            if (count <= 0) {
+                stringRedisTemplate.opsForHash().put(controllerId + "_cmd:" + cmdtopicsMap.getName(), "path", cmdtopicsMap.getPath());
                 stringRedisTemplate.persist(controllerId);
                 sysControllerService.insertsyscontrollercmd(controllerId, timestamp, fleetId, cmdtopicsMap.getName(), cmdtopicsMap.getPath());
                 controllercountcmdcount++;
             }
         }
-        Integer count =sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
-        if (count<=0){
-            stringRedisTemplate.opsForHash().put(controllerId+"_fault:"+faultprot.getName(), "path", faultprot.getPath());
+
+        Integer count = sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
+        if (count <= 0) {
+            stringRedisTemplate.opsForHash().put(controllerId + "_fault:" + faultprot.getName(), "path", faultprot.getPath());
             sysControllerService.insertsyscontrollerfault(controllerId, timestamp, fleetId, faultprot.getName(), faultprot.getPath());
             stringRedisTemplate.persist(controllerId);
         }
-        //如果控制器中的数据出现更新情况，就触发该方法进行重连
-        if (controllercountcount>0) {
-            //关闭连接线程
+
+        if (controllercountcount > 0) {
             messageListenerService2.destroy();
-            //重新建立连接
             messageListenerService2.initMqttConnection();
         }
     }