增加节点保存删除已加载实例功能

byit-myth-admin/src/main/java/com/byit/service/ApiNodeCheckService.java

@@ -9,10 +9,6 @@ public interface ApiNodeCheckService {
      * 工作流校验
      * @param mainFlowId 主工作流的id
      */
-    void checkFlowNode(Integer mainFlowId);
+    int checkFlowNode(Integer mainFlowId);
 
-    /**
-     * 工作流节点校验 删除不在调度上的节点
-     */
-    int checkFlowNode();
 }

byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java

@@ -133,7 +133,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         log.debug("生成版本");
         FlowVersion flowVersion = saveFlowVersion(flow);
 
-        int checkFlowNode = apiNodeCheckService.checkFlowNode();
+        int checkFlowNode = apiNodeCheckService.checkFlowNode(flow.getFlowId());
         log.warn("-------删除错误节点{}个--------",checkFlowNode);
         log.debug("插件端通过API保存成功！");
     }

byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeCheckServiceImpl.java

 package com.byit.service.impl;
 
-import com.byit.model.Node;
-import com.byit.model.vo.FlowVo;
-import com.byit.service.ApiNodeCheckService;
-import com.byit.service.FlowService;
-import com.byit.service.NodeService;
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.enums.DagCheckEnum;
+import com.byit.model.RunRecording;
+import com.byit.service.*;
+import com.byit.utils.ValidationUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
 
 /**
  * @author huangfu
@@ -21,54 +19,44 @@ import java.util.stream.Collectors;
 @Slf4j
 public class ApiNodeCheckServiceImpl implements ApiNodeCheckService {
     private final NodeService nodeService;
-    private final FlowService flowService;
+    private final JobTaskService jobTaskService;
+    private final RunRecordingService runRecordingService;
 
-    public ApiNodeCheckServiceImpl(NodeService nodeService, FlowService flowService) {
+    public ApiNodeCheckServiceImpl(NodeService nodeService, JobTaskService jobTaskService, RunRecordingService runRecordingService) {
         this.nodeService = nodeService;
-        this.flowService = flowService;
+        this.jobTaskService = jobTaskService;
+        this.runRecordingService = runRecordingService;
     }
 
     /**
-     * 废弃原因：查询次数多，性能低
      * @param mainFlowId 主工作流的id
      */
     @Override
-    @Deprecated
-    public void checkFlowNode(Integer mainFlowId) {
-        //查询工作流
-        FlowVo mainFlow = flowService.getFlowId(mainFlowId);
-        String flowVersionName = mainFlow.getVersionName();
-        //基于主工作流查询所有的节点
-        List<Node> mainNode = nodeService.findNodeByFlowIdAndVersionName(mainFlowId);
-        //迭代获取所有的虚节点
-        List<Node> allNodes = new ArrayList<>();
-        getAllInnerNode(mainNode,allNodes);
-        //筛选版本号不一致的
-        List<Node> errorNode = allNodes.stream().filter(node -> {
-            String nodeVersionName = node.getVersionName();
-            return !flowVersionName.equals(nodeVersionName);
-        }).collect(Collectors.toList());
-        List<Integer> collect = errorNode.stream().map(Node::getNodeId).collect(Collectors.toList());
-        log.warn("-------删除错误节点{}个--------",collect.size());
-        nodeService.deleteNode(collect);
-    }
+    public int checkFlowNode(Integer mainFlowId) {
+        //查询实例表
+        List<RunRecording> runRecordingByFlowId = runRecordingService.findRunRecordingByFlowId(mainFlowId);
+        log.debug("--------{}下有{}个实例------------",mainFlowId,runRecordingByFlowId.size());
+        //检查是否有运行中的
+        boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(mainFlowId);
+        ValidationUtil.isTrueValidation(runRecordingIsRunning, "当前工作流正在运行不允许重新发布！");
+        if(CollectionUtil.isNotEmpty(runRecordingByFlowId)){
+            RunRecording runRecording = runRecordingByFlowId.stream().filter(recording -> {
+                boolean isFlow = "1".equals(recording.getFlowStatus());
+                boolean isType = 1 == recording.getScheduleType();
+                return isFlow && isType;
+            }).findFirst().orElse(null);
+            if(runRecording != null) {
+                String runId = runRecording.getRunId();
+                //根据runID删除实例
+                int deleteRunId = runRecordingService.deleteRunId(runId);
+                log.info("-----删除已经被加载的实例{}个------",deleteRunId);
+                //根据runID删除任务表
+                int removeByRunId = jobTaskService.removeByRunId(runId);
+                log.info("-----删除已经被加载的任务{}个------",removeByRunId);
+            }
+        }
 
-    /**
-     * 删除不在调度上的节点
-     */
-    @Override
-    public int checkFlowNode() {
         return nodeService.deleteNoForkNode();
     }
 
-    private void getAllInnerNode(List<Node> mainNodes,List<Node> candidateNodes){
-        for (Node node : mainNodes) {
-            if("1".equals(node.getIsVirtual())) {
-                candidateNodes.add(node);
-            }else{
-                List<Node> mainNode = nodeService.findNodeByFlowIdAndVersionName(node.getMapFlowId());
-                getAllInnerNode(mainNode,candidateNodes);
-            }
-        }
-    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskService.java

@@ -58,5 +58,5 @@ public interface JobTaskService {
 
     void deleteInIds(List<Integer> ids);
 
-    void removeByRunId(String runId);
+    int removeByRunId(String runId);
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunRecordingService.java

@@ -68,6 +68,13 @@ public interface RunRecordingService {
     boolean findRunRecordingIsRunning(Integer flowId);
 
     /**
+     * 工作流ID
+     * @param flowId 工作流ID
+     * @return 该工作流下的所由实例
+     */
+    List<RunRecording> findRunRecordingByFlowId(Integer flowId);
+
+    /**
      * 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
      * @param flowId
      * @param runId
@@ -108,5 +115,7 @@ public interface RunRecordingService {
      */
     int deleteById(Integer recordingId);
 
+    int deleteRunId(String runId);
+
 
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskServiceImpl.java

@@ -91,7 +91,7 @@ public class JobTaskServiceImpl implements JobTaskService {
     }
 
     @Override
-    public void removeByRunId(String runId) {
-        jobTaskMapper.deleteByRunId(runId);
+    public int removeByRunId(String runId) {
+        return jobTaskMapper.deleteByRunId(runId);
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunRecordingServiceImpl.java

 package com.byit.service.impl;
 
+import cn.hutool.core.collection.CollectionUtil;
 import com.byit.dto.FlowConditionDto;
 import com.byit.dto.StatisticsConditionDto;
 import com.byit.enums.RunRecordingEnum;
@@ -106,18 +107,25 @@ public class RunRecordingServiceImpl implements RunRecordingService  {
     @Override
     public boolean findRunRecordingIsRunning(Integer flowId) {
         List<RunRecording> runRecordings = runRecordingMapper.findRunRecordingByFlowId(flowId);
-        for (RunRecording runRecording : runRecordings) {
-            //判断当前的实例  中是否有 未完成的
-            if (RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode().equals(runRecording.getFlowStatus()) ||
-                    RunRecordingEnum.FLOW_STATUS_IS_STOP.getCode().equals(runRecording.getFlowStatus())) {
-                log.debug("----------{}实例正在运行中--------",runRecording);
-                return true;
-            }
-        }
+       if(CollectionUtil.isNotEmpty(runRecordings)) {
+           for (RunRecording runRecording : runRecordings) {
+               //判断当前的实例  中是否有 未完成的
+               if (RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode().equals(runRecording.getFlowStatus()) ||
+                       RunRecordingEnum.FLOW_STATUS_IS_STOP.getCode().equals(runRecording.getFlowStatus())) {
+                   log.debug("----------{}实例正在运行中--------",runRecording);
+                   return true;
+               }
+           }
+       }
         return false;
     }
 
     @Override
+    public List<RunRecording> findRunRecordingByFlowId(Integer flowId) {
+        return runRecordingMapper.findRunRecordingByFlowId(flowId);
+    }
+
+    @Override
     public RunRecording findRunRecordingByFlowIdAndRunId(Integer flowId, String runId) {
         return runRecordingMapper.findRunRecordingByFlowIdAndRunId(flowId,runId);
     }
@@ -146,4 +154,9 @@ public class RunRecordingServiceImpl implements RunRecordingService  {
     public int deleteById(Integer recordingId) {
         return runRecordingMapper.deleteById(recordingId);
     }
+
+    @Override
+    public int deleteRunId(String runId) {
+        return runRecordingMapper.deleteByRunId(runId);
+    }
 }