更新！

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/controller/free/GateWayFreeController.java

@@ -7,6 +7,7 @@ import com.zhongshu.iot.client.model.mqtt.MqttInfoReturnModel;
 import com.zhongshu.iot.server.core.service.base.AkSignService;
 import com.zhongshu.iot.server.core.service.mqtt.DeviceInfoService;
 import com.zhongshu.iot.server.core.service.mqtt.GateWayInfoService;
+import com.zhongshu.iot.server.core.service.mqtt.GateWayUserInfoService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.extern.slf4j.Slf4j;
@@ -40,6 +41,9 @@ public class GateWayFreeController {
     @Autowired
     private AkSignService akSignService;
 
+    @Autowired
+    private GateWayUserInfoService gateWayUserInfoService;
+
     @Operation(summary = "注册网关")
     @RequestMapping(value = "registerGateWay", method = {RequestMethod.POST})
     public ResultContent<MqttInfoReturnModel> registerGateWay(@RequestBody GateWayInfoAddParam param) {

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/dataConfig/MqttConfig.java

@@ -0,0 +1,11 @@
+package com.zhongshu.iot.server.core.dataConfig;
+
+/**
+ * @author TRX
+ * @date 2025/1/2
+ */
+public class MqttConfig {
+
+    public static final String adminRoleName = "amq";
+
+}

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/init/MqttInfoInit.java

@@ -0,0 +1,44 @@
+package com.zhongshu.iot.server.core.init;
+
+import com.zhongshu.iot.server.core.service.base.InitService;
+import com.zhongshu.iot.server.core.service.mqtt.GateWayUserInfoService;
+import com.zhongshu.iot.server.core.service.mqtt.MqttInfoService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author TRX
+ * @date 2025/1/2
+ */
+@Component
+@Slf4j
+public class MqttInfoInit implements CommandLineRunner {
+
+    @Autowired
+    private MqttInfoService mqttInfoService;
+
+    @Autowired
+    private GateWayUserInfoService gateWayUserInfoService;
+
+    @Override
+    public void run(String... args) throws Exception {
+        CompletableFuture.runAsync(() -> {
+            try {
+                TimeUnit.SECONDS.sleep(3);
+
+                // 初始MQTT 连接路径信息
+                mqttInfoService.initMqttInfo();
+
+                // 刷新所有用户和用户权限数据
+                gateWayUserInfoService.initData();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        });
+    }
+}

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/listener/MongodbAutoEvent.java

@@ -42,7 +42,7 @@ public class MongodbAutoEvent extends AbstractMongoEventListener<Object> {
             String className = source.getClass().getName();
             log.info("onAfterDelete 删除了数据... {}", event.getType().getName());
             Document document = event.getDocument();
-            if (document != null) {
+            if (document != null && document.getObjectId("_id") != null) {
                 String id = document.getObjectId("_id").toString();
                 if (className.equals(DeviceOnLineInfo.class.getName())) {
                     devicePingInfoService.deviceTTLUnLine(id);

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/service/base/InitService.java

@@ -1,8 +1,8 @@
 package com.zhongshu.iot.server.core.service.base;
 
+import com.github.microservice.net.ResultContent;
 import com.zhongshu.iot.server.core.service.mqtt.GateWayUserInfoService;
 import com.zhongshu.iot.server.core.service.user.impl.UserServiceImpl;
-import com.github.microservice.net.ResultContent;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -16,13 +16,14 @@ import org.springframework.stereotype.Service;
 public class InitService extends SuperService {
 
     @Autowired
-    UserServiceImpl userService;
+    private UserServiceImpl userService;
 
     @Autowired
-    GateWayUserInfoService gateWayUserInfoService;
+    private GateWayUserInfoService gateWayUserInfoService;
 
     public ResultContent init() {
-        userService.initAdmin();
+        // 初始平台用户信息
+//        userService.initAdmin();
 
         gateWayUserInfoService.initDefaultUser();
 

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/service/mqtt/GateWayUserInfoService.java

@@ -7,6 +7,7 @@ import com.zhongshu.iot.client.model.mqtt.GateWayUserInfoNameParam;
 import com.zhongshu.iot.client.model.mqtt.GateWayUserInfoSearchParam;
 import com.zhongshu.iot.client.type.type.MqttUserState;
 import com.zhongshu.iot.server.core.dao.mqtt.*;
+import com.zhongshu.iot.server.core.dataConfig.MqttConfig;
 import com.zhongshu.iot.server.core.domain.iot.mqtt.*;
 import com.zhongshu.iot.server.core.util.CommonUtil;
 import com.zhongshu.iot.server.core.util.DateUtils;
@@ -56,9 +57,26 @@ public class GateWayUserInfoService {
 
     @Autowired
     private GateWayMqttSecurityDao gateWayMqttSecurityDao;
+
     @Autowired
     private GateWay2DeviceDao gateWay2DeviceDao;
 
+    @Autowired
+    private GateWayInfoDao gateWayInfoDao;
+
+    /**
+     * 刷新系统所有的用户 和 用户权限数据
+     */
+    public void initData() {
+        log.info("GateWayUserInfoService initData");
+
+        initDefaultUser();
+
+        refreshAllGateWayUser();
+
+        refreshAllGateWaySecurity();
+    }
+
     /**
      * 初始mqtt连接账号信息
      *
@@ -69,8 +87,9 @@ public class GateWayUserInfoService {
         GateWayUserInfo entity = gateWayUserInfoDao.findTopByUserName(userName);
         if (ObjectUtils.isEmpty(entity)) {
             GateWayUserInfoAddParam param = new GateWayUserInfoAddParam();
-            param.setUserName("admin");
+            param.setUserName(userName);
             param.setPassWord("admin123");
+            param.setRoleName(MqttConfig.adminRoleName);
             addGateWayUser(param);
             log.info(String.format("gateWayUser [%s] 初始化成功", userName));
         } else {
@@ -234,6 +253,18 @@ public class GateWayUserInfoService {
         return ResultContent.buildSuccess();
     }
 
+    public ResultContent syncUserToMQTTService(List<GateWayUserInfo> gateWayUserInfoList) {
+        //todo同步用户
+        List<MqttInfo> list = mqttInfoDao.findAll();
+        log.info("syncUserToMQTTService {}", list.size());
+        if (ObjectUtils.isNotEmpty(list)) {
+            list.parallelStream().forEach(mqttInfo -> {
+                syncMqttUsers(mqttInfo, gateWayUserInfoList);
+            });
+        }
+        return ResultContent.buildSuccess();
+    }
+
     /**
      * 同步多个用户
      *
@@ -338,6 +369,18 @@ public class GateWayUserInfoService {
         return ResultContent.buildSuccess();
     }
 
+    public ResultContent syncSecurityToMQTTService(List<GateWayInfo> gateWayInfos) {
+        //todo同步权限
+        List<MqttInfo> list = mqttInfoDao.findAll();
+        log.info("syncSecurityToMQTTService {}", list.size());
+        if (ObjectUtils.isNotEmpty(list)) {
+            list.parallelStream().forEach(mqttInfo -> {
+                syncMqttSecuritySettings(mqttInfo, gateWayInfos);
+            });
+        }
+        return ResultContent.buildSuccess();
+    }
+
     /**
      * 同步网关可监听的权限
      *
@@ -393,11 +436,11 @@ public class GateWayUserInfoService {
                             gateWayMqttSecurity.setGateWayId(gateWayInfo.getGateWayId());
                             gateWayMqttSecurity.setMqttInfo(mqttInfo);
                             gateWayMqttSecurity.setMqttName(mqttInfo.getName());
+                            gateWayMqttSecurity.setAddressMatch(addressMatch);
                             try {
-                                addressControl.addSecuritySettings(
-                                        addressMatch,
-                                        roleName, roleName, roleName, roleName, roleName, roleName, roleName,
-                                        roleName, roleName, roleName);
+                                if (!roleName.equals(MqttConfig.adminRoleName)) {
+                                    addressControl.addSecuritySettings(addressMatch, roleName, roleName, roleName, roleName, roleName, roleName, roleName, roleName, roleName, roleName);
+                                }
                                 gateWayMqttSecurity.setIsSync(Boolean.TRUE);
                                 gateWayMqttSecurity.setMsg("同步成功");
                             } catch (Exception e) {
@@ -417,6 +460,32 @@ public class GateWayUserInfoService {
         }
     }
 
+    /**
+     * 同步所有的用户
+     *
+     * @return
+     */
+    public ResultContent refreshAllGateWayUser() {
+        List<GateWayUserInfo> gateWayUserInfoList = gateWayUserInfoDao.findAll();
+        if (ObjectUtils.isNotEmpty(gateWayUserInfoList)) {
+            syncUserToMQTTService(gateWayUserInfoList);
+        }
+        return ResultContent.buildSuccess();
+    }
+
+    /**
+     * 同步所有的数据
+     *
+     * @return
+     */
+    public ResultContent refreshAllGateWaySecurity() {
+        List<GateWayInfo> gateWayInfos = gateWayInfoDao.findAll();
+        if (ObjectUtils.isNotEmpty(gateWayInfos)) {
+            syncSecurityToMQTTService(gateWayInfos);
+        }
+        return ResultContent.buildSuccess();
+    }
+
     //-----------------------------同步权限数据 end----------------------
 
     /**

OneCardIotServer/src/main/java/com/zhongshu/iot/server/core/service/mqtt/MqttInfoService.java

@@ -19,7 +19,6 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.PostConstruct;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -44,24 +43,23 @@ public class MqttInfoService {
     /**
      * 初始数据
      */
-    @PostConstruct
     public void initMqttInfo() {
         log.info("PostConstruct initMqttInfo");
         MqttInfo mqttInfo = mqttInfoDao.findTopByAddress(AddressType.Common);
         if (ObjectUtils.isEmpty(mqttInfo)) {
             mqttInfo = new MqttInfo();
-            mqttInfo.setName("默认MQTT服务");
-            mqttInfo.setBrokerHost("wss://mq.dev.zswllife.cn");
-            mqttInfo.setBrokerPort("443/mqtt");
-            mqttInfo.setUserName("admin");
-            mqttInfo.setPassword("admin123");
-            mqttInfo.setAddress(AddressType.Common);
-            mqttInfo.setState(CommonState.Enable);
-            mqttInfo.setBrokerName("broker1");
-            mqttInfo.setJmxHost("172.24.50.53");
-            mqttInfo.setJmxPort("1098");
-            mqttInfoDao.save(mqttInfo);
         }
+        mqttInfo.setName("默认MQTT服务");
+        mqttInfo.setBrokerHost("wss://mq.dev.zswllife.cn");
+        mqttInfo.setBrokerPort("443/mqtt");
+        mqttInfo.setUserName("admin");
+        mqttInfo.setPassword("admin123");
+        mqttInfo.setAddress(AddressType.Common);
+        mqttInfo.setState(CommonState.Enable);
+        mqttInfo.setBrokerName("broke1");
+        mqttInfo.setJmxHost("172.24.50.53");
+        mqttInfo.setJmxPort("1098");
+        mqttInfoDao.save(mqttInfo);
     }
 
     /**
@@ -77,8 +75,8 @@ public class MqttInfoService {
         GateWayUserInfo gateWayUserInfo = gateWayUserInfoService.gateWayBindMqttUser(gatewayInfo);
         if (ObjectUtils.isNotEmpty(mqttInfo)) {
             mqttInfoSimpleModel.setBrokerAddress(String.format("%s:%s", mqttInfo.getBrokerHost(), mqttInfo.getBrokerPort()));
-            mqttInfoSimpleModel.setBrokerUsername(gateWayUserInfo.getUserName());
-            mqttInfoSimpleModel.setBrokerPassword(gateWayUserInfo.getPassWord());
+            mqttInfoSimpleModel.setBrokerUsername(mqttInfo.getUserName());
+            mqttInfoSimpleModel.setBrokerPassword(mqttInfo.getPassword());
         }
         return mqttInfoSimpleModel;
     }