202209061653

gather-broker/src/main/java/com/idtgz/model/pipeline/ConnectDTO.java

@@ -45,7 +45,7 @@ public class ConnectDTO {
 
     public int status;
 
-    public Boolean isUpload;
+    public int isUpload;
 
     List<String> items;
 
@@ -71,7 +71,7 @@ public class ConnectDTO {
         this.device = connectionEntity.getDeviceId();
         this.mode = connectionEntity.getConnectionProtocol();
         this.updateTimestamp = connectionEntity.getUpdatedTime().toEpochSecond(ZoneOffset.of("+8")) * 1000;
-        this.isUpload = connectionEntity.getIsUploaded() == 1 ? true : false;
+        this.isUpload = connectionEntity.getIsUploaded();
         this.status = connectionEntity.getConnectionStatus();
         this.clientId = connectionEntity.getClientId();
         this.storeStrategy = connectionEntity.getStoreStrategy();

gather-broker/src/main/java/com/idtgz/mqtt/api/impl/AuthServiceImpl.java

@@ -66,7 +66,7 @@ public class AuthServiceImpl implements MqttAuth {
                         connectionEntity.setIsUploaded(UploadStatus.UPLOADING);
                         connectionEntity.setConnectionStatus(ConfigConstant.CONNECTION_STATUS_ON);
                         connectionMapper.updateById(connectionEntity);
-                        connectionService.updateSchedule(connectionEntity);
+                        connectionService.updateSchedule(connectionEntity,false);
                     }
                     connectionMapper.updateById(connectionEntity);
                     ConfigContext.connections.put(connectionEntity.getDeviceId(), connectionEntity);

gather-broker/src/main/java/com/idtgz/mqtt/protocol/ProtocolProcess.java

@@ -207,7 +207,7 @@ public class ProtocolProcess implements InternalRecvice {
     public void processDisConnect(Channel channel, MqttMessage msg) {
         String clientId = NettyUtil.getClientId(channel);
         if (clientId.startsWith(ConfigConstant.CLIENT_PREFIX)) {
-            clientService.clientDisconnect(clientId.replace(ConfigConstant.CLIENT_PREFIX, ""));
+            clientService.clientDisconnect(clientId);
         } else if (clientId.startsWith(ConfigConstant.MQTT_CONNECTION_PREFIX)) {
             clientService.clientDisconnect(clientId);
         }

gather-broker/src/main/java/com/idtgz/service/ConnectionService.java

@@ -65,7 +65,9 @@ public interface ConnectionService extends IService<ConnectionEntity> {
     //更新上传任务
     Boolean updateSchedule(String connectionSign, int uploadType, int uploadcycle, Integer storeStrategy, Integer status, boolean throwEx) throws SchedulerException;
 
-    Boolean updateSchedule(ConnectionEntity connectionEntity) throws SchedulerException;
+    Boolean updateSchedule(Long deviceId, int uploadType, int uploadcycle, Integer storeStrategy, Integer status, boolean throwEx) throws SchedulerException;
+
+    Boolean updateSchedule(ConnectionEntity connectionEntity, boolean throwEx) throws SchedulerException;
 
     //暂停上传任务
     Boolean stopSchedule(Long deviceId);

gather-broker/src/main/java/com/idtgz/service/impl/ClientServiceImpl.java

@@ -78,6 +78,7 @@ public class ClientServiceImpl implements ClientService {
     public void clientDisconnect(String clientId) {
         //级联客户端断连
         if (clientId.startsWith(ConfigConstant.CLIENT_PREFIX)) {
+            clientId = clientId.replace(ConfigConstant.CLIENT_PREFIX,"");
             ClientEntity clientEntity = ConfigContext.clients.get(clientId);
             clientEntity.setClientStatus(ConfigConstant.CLIENT_STATUS_OFF);
             clientMapper.updateById(clientEntity);

gather-broker/src/main/java/com/idtgz/service/impl/ConnectionServiceImpl.java

@@ -531,14 +531,42 @@ public class ConnectionServiceImpl extends ServiceImpl<ConnectionMapper, Connect
         connectionEntity.setUploadCycle(uploadcycle);
         connectionEntity.setIsUploaded(status);
         connectionEntity.setStoreStrategy(storeStrategy);
+        //内部configJosn更新
+        if (connectionEntity.getConnectionProtocol() == ProtocolConstant.OPCDA) {
+            OPCDaConfig opcDaConfig = JSON.parseObject(connectionEntity.getConnectionJson(), OPCDaConfig.class);
+            opcDaConfig.setUploadType(uploadType);
+            opcDaConfig.setCycle(uploadcycle);
+            connectionEntity.setConnectionJson(JSON.toJSONString(opcDaConfig));
+        } else if (connectionEntity.getConnectionProtocol() == ProtocolConstant.IEC104) {
+            IEC104Config iec104Config = JSON.parseObject(connectionEntity.getConnectionJson(), IEC104Config.class);
+            iec104Config.setUploadType(uploadType);
+            iec104Config.setCycle(uploadcycle);
+            connectionEntity.setConnectionJson(JSON.toJSONString(iec104Config));
+        } else if (connectionEntity.getConnectionProtocol() == ProtocolConstant.MQTT) {
+            MqttConnectionConfig mqttConnectionConfig = JSON.parseObject(connectionEntity.getConnectionJson(), MqttConnectionConfig.class);
+            mqttConnectionConfig.setUploadType(uploadType);
+            mqttConnectionConfig.setCycle(uploadcycle);
+            connectionEntity.setConnectionJson(JSON.toJSONString(mqttConnectionConfig));
+        }
         this.updateById(connectionEntity);
         ConfigContext.connections.put(connectionEntity.getDeviceId(), connectionEntity);
+        centerService.remoteSyn();
         return true;
     }
 
+
     @Override
-    public Boolean updateSchedule(ConnectionEntity connectionEntity) throws SchedulerException {
+    public Boolean updateSchedule(Long deviceId, int uploadType, int uploadcycle, Integer storeStrategy, Integer status, boolean throwEx) throws SchedulerException {
-        return updateSchedule(connectionEntity.getConnectionSign(), connectionEntity.getUploadType(), connectionEntity.getUploadCycle(), connectionEntity.getStoreStrategy(), connectionEntity.getIsUploaded(), false);
+        ConnectionEntity connectionEntity = this.getOne(new QueryWrapper<ConnectionEntity>().eq("device_id", deviceId));
+        connectionEntity.setUploadType(uploadType);
+        connectionEntity.setUploadCycle(uploadcycle);
+        connectionEntity.setIsUploaded(status);
+        return updateSchedule(connectionEntity, throwEx);
+    }
+
+    @Override
+    public Boolean updateSchedule(ConnectionEntity connectionEntity, boolean throwEx) throws SchedulerException {
+        return updateSchedule(connectionEntity.getConnectionSign(), connectionEntity.getUploadType(), connectionEntity.getUploadCycle(), connectionEntity.getStoreStrategy(), connectionEntity.getIsUploaded(), throwEx);
     }
 
     @Override
@@ -625,23 +653,20 @@ public class ConnectionServiceImpl extends ServiceImpl<ConnectionMapper, Connect
                 isSuccess = this.delBrokerConnection(connectDTO.getConnectSign());
                 break;
         }
-        mqttClientService.deviceResponse(request.getMethod(), request.getRequestId(), isSuccess, "", "{}");
+        mqttClientService.deviceResponse(request.getMethod(), request.getRequestId(), isSuccess, "", connectDTO);
         this.syncDeviceChannelConfig();
     }
 
     @Override
     public void updateConectionConfig(DeviceBaseRequest request) {
-        List<ConnectDTO> list = JSON.parseObject(request.getPayload().toString(), List.class);
+        ConnectDTO connectDTO = JSON.parseObject(request.getPayload().toString(), ConnectDTO.class);
-        list.forEach(
-                i -> {
         try {
-                        updateSchedule(i.getConnectSign(), i.getUploadType(), i.getCycle(), i.getStoreStrategy(), i.getStatus(), false);
+            updateSchedule(connectDTO.getDevice(), connectDTO.getUploadType(), connectDTO.getCycle(), connectDTO.getStoreStrategy(), connectDTO.getStatus(), true);
-                    } catch (SchedulerException e) {
+        } catch (Exception e) {
-                        e.printStackTrace();
+            mqttClientService.deviceResponse(request.getMethod(), request.getRequestId(), false, e.getMessage(), connectDTO);
+            return;
         }
-                }
+        mqttClientService.deviceResponse(request.getMethod(), request.getRequestId(), true, null, connectDTO);
-        );
-        mqttClientService.deviceResponse(request.getMethod(), request.getRequestId(), true, null, null);
     }
 
     @Override

gather-broker/src/main/java/com/idtgz/service/impl/MqttClientServiceImpl.java

@@ -153,7 +153,7 @@ public class MqttClientServiceImpl implements MqttClientService {
         DeviceBaseResponse response = new DeviceBaseResponse();
         response.requestId = requestId;
         response.method = method;
-        response.payload = o == null ? "{}" : o;
+        response.payload = o == null ? (new Object()) : o;
         response.success = isSuccess;
         response.error = err;
         MqttMessage mqttMessage = new MqttMessage();