fix:修复iot采集设备问题

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicemodelattribute/DeviceModelAttributeMapper.java

@@ -47,7 +47,7 @@ public interface DeviceModelAttributeMapper extends BaseMapperX<DeviceModelAttri
                 .eqIfPresent(DeviceModelAttributeDO::getDataUnit, reqVO.getDataUnit())
                 .eqIfPresent(DeviceModelAttributeDO::getRatio, reqVO.getRatio())
                 .eqIfPresent(DeviceModelAttributeDO::getRemark, reqVO.getRemark())
-                .eqIfPresent(DeviceModelAttributeDO::getDeviceModelId, reqVO.getId())
+                .eqIfPresent(DeviceModelAttributeDO::getDeviceModelId, reqVO.getDeviceModelId())
                 .betweenIfPresent(DeviceModelAttributeDO::getCreateTime, reqVO.getCreateTime())
                 .orderByDesc(DeviceModelAttributeDO::getId);
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceServiceImpl.java

@@ -47,6 +47,7 @@ import cn.iocoder.yudao.module.iot.dal.mysql.mqttdatarecord.MqttDataRecordMapper
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.DeviceAttributeDO;
 import cn.iocoder.yudao.module.iot.dal.mysql.device.DeviceAttributeMapper;
 
+import cn.iocoder.yudao.module.iot.framework.mqtt.config.DefaultBizTopicSet;
 import cn.iocoder.yudao.module.iot.framework.mqtt.consumer.IMqttservice;
 import cn.iocoder.yudao.module.iot.util.MapListStatsCalculator;
 import com.alibaba.fastjson.JSON;
@@ -139,6 +140,9 @@ public class DeviceServiceImpl implements DeviceService {
     @Resource
     private IMqttservice mqttService;
 
+    @Resource
+    private DefaultBizTopicSet defaultBizTopicSet;
+
     @Resource
     private ErpProductUnitMapper productUnitMapper;
 
@@ -273,24 +277,59 @@ public class DeviceServiceImpl implements DeviceService {
     }
 
     @Override
+    @Transactional(rollbackFor = Exception.class)
     public void updateDevice(DeviceSaveReqVO updateReqVO) {
         // 校验存在
         validateDeviceExists(updateReqVO.getId());
         //校驗topic是否唯一
-        validTopicExists(updateReqVO.getTopic());
+        validTopicExists(updateReqVO.getTopic(), updateReqVO.getId());
 
         // 更新
         DeviceDO updateObj = BeanUtils.toBean(updateReqVO, DeviceDO.class);
         deviceMapper.updateById(updateObj);
     }
 
-    private void validTopicExists(String topic) {
+    private void validTopicExists(String topic, Long currentDeviceId) {
-        boolean exists = deviceMapper.exists(Wrappers.<DeviceDO>lambdaQuery().eq(DeviceDO::getTopic, topic));
+        if (StringUtils.isBlank(topic)) {
+            return;
+        }
+        boolean exists = deviceMapper.exists(Wrappers.<DeviceDO>lambdaQuery()
+                .eq(DeviceDO::getTopic, topic)
+                .ne(currentDeviceId != null, DeviceDO::getId, currentDeviceId));
         if (exists){
             throw exception(DEVICE_MQTT_TOPIC_ALREADY_EXIST);
         }
     }
 
+    private void disableGatewayByDevice(DeviceDO device) {
+        if (device == null) {
+            return;
+        }
+        LambdaUpdateWrapper<GatewayDO> updateWrapper = new LambdaUpdateWrapper<GatewayDO>()
+                .set(GatewayDO::getIsEnable, false)
+                .set(GatewayDO::getUpdateTime, LocalDateTime.now());
+        if (StringUtils.isNotBlank(device.getTopic())) {
+            updateWrapper.eq(GatewayDO::getTopic, device.getTopic());
+        } else if (StringUtils.isNotBlank(device.getDeviceCode())) {
+            updateWrapper.eq(GatewayDO::getGatewayCode, device.getDeviceCode());
+        } else {
+            return;
+        }
+        if (StringUtils.isNotBlank(device.getDeviceCode())) {
+            updateWrapper.or().eq(GatewayDO::getGatewayCode, device.getDeviceCode());
+        }
+        gatewayMapper.update(null, updateWrapper);
+    }
+
+    private void removeRuntimeTopic(String topic) {
+        if (StringUtils.isBlank(topic) || defaultBizTopicSet == null || defaultBizTopicSet.getTopicMap() == null) {
+            return;
+        }
+        synchronized (defaultBizTopicSet.getTopicMap()) {
+            defaultBizTopicSet.getTopicMap().removeIf(item -> topic.equals(item.getSubTopic()));
+        }
+    }
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void deleteDevice(List<Long> ids) {
@@ -323,10 +362,12 @@ public class DeviceServiceImpl implements DeviceService {
         deviceContactModelMapper.delete(deviceContactModelDOLambdaQueryWrapper);
     }
 
-    private void validateDeviceExists(Long id) {
+    private DeviceDO validateDeviceExists(Long id) {
-        if (deviceMapper.selectById(id) == null) {
+        DeviceDO device = deviceMapper.selectById(id);
+        if (device == null) {
             throw exception(DEVICE_NOT_EXISTS);
         }
+        return device;
     }
 
     @Override
@@ -795,6 +836,7 @@ public class DeviceServiceImpl implements DeviceService {
     }
 
     @Override
+//    @Transactional(rollbackFor = Exception.class)
     public Long copyDevice(Long id) {
         if (id == null){
             throw exception(DEVICE_ID_MODEL_NOT_EXISTS);
@@ -811,6 +853,7 @@ public class DeviceServiceImpl implements DeviceService {
         int randomNumber = random.nextInt(9000) + 1000;
         newDevice.setDeviceCode(deviceDO.getDeviceCode()+ "-" + randomNumber);
         newDevice.setDeviceName(deviceDO.getDeviceName() + "-副本");
+        newDevice.setTopic(null);
         deviceMapper.insert(newDevice);
 
         //复制关联表
@@ -818,14 +861,20 @@ public class DeviceServiceImpl implements DeviceService {
         lambdaQueryWrapper.eq(DeviceContactModelDO::getDeviceId,deviceDO.getId());
         List<DeviceContactModelDO> deviceContactModelDOS = deviceContactModelMapper.selectList(lambdaQueryWrapper);
 
-
+        List<DeviceContactModelDO> newContactModelList = new ArrayList<>();
         if (deviceContactModelDOS != null && !deviceContactModelDOS.isEmpty()){
-            for (DeviceContactModelDO deviceModelAttributeDOS : deviceContactModelDOS) {
+            for (DeviceContactModelDO deviceContactModelDO : deviceContactModelDOS) {
-                deviceModelAttributeDOS.setId(null);
+                DeviceContactModelDO newContactModel = new DeviceContactModelDO();
-                deviceModelAttributeDOS.setDeviceId(newDevice.getId());
+                BeanUtils.copyProperties(deviceContactModelDO, newContactModel);
+                newContactModel.setId(null);
+                newContactModel.setDeviceId(newDevice.getId());
+                newContactModel.setCreateTime(LocalDateTime.now());
+                newContactModel.setUpdateTime(LocalDateTime.now());
+                newContactModelList.add(newContactModel);
             }
-            deviceContactModelMapper.insertBatch(deviceContactModelDOS);
+            deviceContactModelMapper.insertBatch(newContactModelList);
         }
+        tdengineService.createTdengineTable(newDevice.getId(), newContactModelList);
 
         return newDevice.getId();
 
@@ -2032,12 +2081,7 @@ public class DeviceServiceImpl implements DeviceService {
             throw exception(DEVICE_MQTT_TOPIC_EXIST);
         }
 
-        // 2. 如果状态没有变化，直接返回
+        // 2. 执行状态同步操作：重复启停也要同步 MQTT 与 gateway，保证幂等
-        if (Objects.equals(deviceDO.getIsEnable(), updateEnabledReqVO.getEnabled())) {
-            return;
-        }
-
-        // 3. 执行状态更新操作
         executeEnableUpdate(deviceDO, updateEnabledReqVO.getEnabled());
 
 
@@ -2051,6 +2095,8 @@ public class DeviceServiceImpl implements DeviceService {
 
     private void executeEnableUpdate(DeviceDO deviceDO, Boolean enabled) {
 
+        MqttException syncException = null;
+
         try {
             // 1. 更新设备启用状态
             deviceDO.setIsEnable(enabled);
@@ -2066,6 +2112,7 @@ public class DeviceServiceImpl implements DeviceService {
                     mqttService.subscribeTopic(topic);
                     log.info("MQTT订阅成功: {}", topic);
                 } catch (MqttException e) {
+                    syncException = e;
                     log.error("MQTT订阅失败: {}", topic, e);
                 }
 
@@ -2091,6 +2138,8 @@ public class DeviceServiceImpl implements DeviceService {
 
                 } else {
                     //更新gateway状态
+                    gateway.setGatewayName(deviceDO.getDeviceName());
+                    gateway.setGatewayCode(deviceDO.getDeviceCode());
                     gateway.setIsEnable(true);
                     gateway.setUpdateTime(LocalDateTime.now());
 
@@ -2109,17 +2158,14 @@ public class DeviceServiceImpl implements DeviceService {
                     mqttService.unsubscribeTopic(topic);
                     log.info("MQTT取消订阅成功: {}", topic);
                 } catch (MqttException e) {
+                    syncException = e;
                     log.error("MQTT取消订阅失败: {}", topic, e);
+                } finally {
+                    removeRuntimeTopic(topic);
                 }
 
                 // 3.2 更新gateway状态为禁用
-                gatewayMapper.update(
+                disableGatewayByDevice(deviceDO);
-                        null,
-                        new LambdaUpdateWrapper<GatewayDO>()
-                                .eq(GatewayDO::getTopic, topic)
-                                .set(GatewayDO::getIsEnable, false)
-                                .set(GatewayDO::getUpdateTime, LocalDateTime.now())
-                );
 
                 log.info("gateway订阅记录已禁用 topic={}", topic);
 
@@ -2130,9 +2176,14 @@ public class DeviceServiceImpl implements DeviceService {
 
             }
 
+            if (syncException != null) {
+                throw new RuntimeException("MQTT订阅状态同步失败", syncException);
+            }
+
         } catch (Exception e) {
 
             log.error("更新设备状态失败 deviceCode={}", deviceDO.getDeviceCode(), e);
+            throw new RuntimeException("更新设备状态失败", e);
 
         }
         }

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java

@@ -779,6 +779,8 @@ public class TDengineService {
                     continue;
                 }
 
+                validateColumnName(attributeCode);
+
                 String tdType = JavaToTdengineTypeEnum.getTdTypeByJavaType(dataType);
                 if (tdType == null) {
                     tdType = "DOUBLE";
@@ -1292,7 +1294,7 @@ public class TDengineService {
         Set<String> reservedWords = new HashSet<>(Arrays.asList(
                 "value", "timestamp", "current", "database", "table",
                 "user", "password", "select", "insert", "update", "delete",
-                "create", "drop", "alter", "show", "describe", "use", "ts"
+                "create", "drop", "alter", "show", "describe", "use", "ts", "state"
         ));
 
         return reservedWords.contains(columnName.toLowerCase());
@@ -1312,7 +1314,7 @@ public class TDengineService {
         Set<String> reservedKeywords = new HashSet<>(Arrays.asList(
                 "value", "timestamp", "current", "database", "table", "user", "password",
                 "select", "insert", "update", "delete", "create", "drop", "alter",
-                "show", "describe", "use", "ts", "now", "current_timestamp"
+                "show", "describe", "use", "ts", "now", "current_timestamp", "state"
         ));
 
         if (reservedKeywords.contains(columnName.toLowerCase())) {
@@ -1359,7 +1361,7 @@ public class TDengineService {
         Set<String> reservedKeywords = new HashSet<>(Arrays.asList(
                 "value", "timestamp", "current", "database", "table", "user", "password",
                 "select", "insert", "update", "delete", "create", "drop", "alter",
-                "show", "describe", "use", "ts", "now", "current_timestamp"
+                "show", "describe", "use", "ts", "now", "current_timestamp", "state"
         ));
 
         if (reservedKeywords.contains(tableName.toLowerCase())) {