取消内嵌节点不允许暂停的设计

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

+package com.byit.service;
+
+import com.byit.dto.operating.StopNodeParam;
+
+/**
+ * 节点操作业务
+ * @author huangfu
+ */
+public interface ApiNodeOperatingService {
+    /**
+     * 暂停节点
+     * 暂停节点不会影响其他非关联节点运行，仅对剩余关联节点起作用
+     * @param stopNodeParam 暂停节点参数
+     */
+    void stopNode(StopNodeParam stopNodeParam);
+}

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

@@ -71,7 +71,6 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
     public void specialRunBatch(SpecialJobParam specialJobParam) {
         //获取操作人
         String operator = currentUserUtils.account();
-        //String operator = "huangfusuper"
         //生成本次补批的运行标识
         String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
         //工作空间名称

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

@@ -1218,6 +1218,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         return true;
     }
 
+    /**
+     * 暂停的是整个工作流
+     * @param runId
+     * @return
+     */
     @Override
     public Boolean stopScheduleByRunId(String runId) {
         List<RunRecording> recordingList = runRecordingMapper.findByRunID(runId);
@@ -1226,7 +1231,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         //"3".equals(runRecording.getFlowStatus()) ||
         recordingList.forEach(runRecording -> {
             if (("4".equals(runRecording.getFlowStatus()))  //如果不是正在运行
-                    && FlowPropertyEnum.ISNOT_INNER.getCode().equals(runRecording.getIsInner())) {       //并且不是内嵌工作流
+                    && FlowPropertyEnum.ISNOT_INNER.getCode().equals(runRecording.getIsInner())) {
                 result.set(false);
             }
         });
@@ -1656,15 +1661,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
         Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
         ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
-        ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许停止调度！");
+        //ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许暂停调度！");
         //判断是否在调度中
         List<RunRecording> recordingList = runRecordingMapper.findUnFinishByFlowId(flow.getFlowId());
         ValidationUtil.isTrueValidation(null == recordingList || recordingList.size() == 0, flowName + "工作流没有正在运行的调度！");
         StringBuffer runids = new StringBuffer();
         //暂停工作流调度
         recordingList.forEach(runRecording -> {
+            //runRecordingMapper.stopByRunIdAndFlowId(runRecording.getRunId(), runRecording.getFlowId());
             runRecordingMapper.stopByRunId(runRecording.getRunId());
-            jobTaskMapper.stopByRunId(runRecording.getRunId());
+            jobTaskMapper.stopByRunIdAndFlowId(runRecording.getRunId(), runRecording.getFlowId());
             runids.append(runRecording.getRunId() + ",");
         });
         return runids.toString();

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

+package com.byit.service.impl;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.dto.NodeRelyDto;
+import com.byit.dto.operating.StopNodeParam;
+import com.byit.model.*;
+import com.byit.service.*;
+import com.byit.util.FlowNodeRelyParseUtil;
+import com.byit.utils.ValidationUtil;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * 节点操作业务
+ *
+ * @author huangfu
+ */
+@Service
+@Transactional(rollbackFor = Exception.class)
+public class ApiNodeOperatingServiceImpl implements ApiNodeOperatingService {
+    /**
+     * 工作流操作
+     */
+    private final FlowService flowService;
+    /**
+     * 工作空间操作
+     */
+    private final WorkspaceService workspaceService;
+    /**
+     * 节点操作
+     */
+    private final NodeService nodeService;
+    /**
+     * 节点依赖操作
+     */
+    private final NodeDependencyService nodeDependencyService;
+
+    private final RunRecordingService runRecordingService;
+    /**
+     * 运行实例操作
+     */
+    private final JobTaskService jobTaskService;
+
+    public ApiNodeOperatingServiceImpl(FlowService flowService, WorkspaceService workspaceService, NodeService nodeService, NodeDependencyService nodeDependencyService, RunRecordingService runRecordingService, JobTaskService jobTaskService) {
+        this.flowService = flowService;
+        this.workspaceService = workspaceService;
+        this.nodeService = nodeService;
+        this.nodeDependencyService = nodeDependencyService;
+        this.runRecordingService = runRecordingService;
+        this.jobTaskService = jobTaskService;
+    }
+
+    @Override
+    public void stopNode(StopNodeParam stopNodeParam) {
+        //工作空间名称
+        String workspaceName = stopNodeParam.getWorkspaceName();
+        ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空");
+        //那个工作流
+        String flowName = stopNodeParam.getFlowName();
+        ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空");
+        //开始节点任务名称
+        String nodeTaskName = stopNodeParam.getNodeName();
+        ValidationUtil.dataNotBank(nodeTaskName, "节点名称不允许为空");
+        Workspace workspace = workspaceService.getByName(workspaceName);
+        ValidationUtil.dataNotNull(workspace, "工作空间不存在！");
+        //查询到工作流
+        Flow flow = flowService.findFlowByName(flowName, workspace.getWorkspaceId());
+        ValidationUtil.dataNotNull(flow, "工作流不存在！");
+        //运行实例
+        RunRecording runRecording = runRecordingService.runIngRunRecording(flowName, workspace.getWorkspaceId());
+        ValidationUtil.dataNotNull(runRecording, "没有运行中的实例！");
+        //查询所有的节点
+        List<Node> flowAllNode = nodeService.findNodeByFlowIdAndVersionName(flow.getFlowId());
+        ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(flowAllNode), String.format("查询到工作流%s下不存在任何节点！请联系调度中心人员！", flowName));
+        //筛选节点信息
+        List<Integer> integerList = flowAllNode.stream().map(Node::getNodeId).collect(Collectors.toList());
+        List<NodeDependencyKey> allNodeDependencyKey = nodeDependencyService.findAllNodeDependencyKey(integerList);
+        //解析对应的版本依赖
+        Set<NodeRelyDto> nodeRelyDtoSet = FlowNodeRelyParseUtil.parseThisVersionNodeRely(flowAllNode, nodeTaskName, allNodeDependencyKey, true);
+        //获取下面所有的节点
+        ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(nodeRelyDtoSet), String.format("查询到工作流%s的%s下不存在任何依赖节点！请联系调度中心人员！", flowName, nodeTaskName));
+        List<Integer> nodeList = nodeRelyDtoSet.stream().map(nodeRelyDto -> nodeRelyDto.getNode().getNodeId()).collect(Collectors.toList());
+        //暂停余下节点信息
+        jobTaskService.stopJobTask(runRecording.getRunId(), nodeList);
+    }
+}

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

@@ -129,7 +129,10 @@ public class ApiNodeServiceImpl implements ApiNodeService {
         if (NodeTypeEnum.JAVA.getCode().equals(runNode.getJobType())) {
             timerTask = new JavaNodeExecutorTask(schedule);
         }else{
-            schedule.setRunParam(PlaceholderUtils.formatParam(schedule.getRunParam()));
+            String privateParam = runParamWrapped.getPrivateParam();
+            String formatParam = PlaceholderUtils.formatParam(privateParam);
+            runParamWrapped.setPrivateParam(formatParam);
+            schedule.setRunParam(JSON.toJSONString(runParamWrapped));
             timerTask = new ScriptExecutorJobTask(schedule);
         }
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java

@@ -91,6 +91,14 @@ public interface JobTaskMapper {
      * @return
      */
     int stopByRunId(String runId);
+
+    /**
+     * 暂停某一确定工作流实例的节点
+     * @param runId
+     * @param flowId
+     * @return
+     */
+    int stopByRunIdAndFlowId(@Param("runId") String runId, @Param("flowId")Integer flowId);
     /**
      * 根据runid开始调度
      * @param runId
@@ -98,4 +106,6 @@ public interface JobTaskMapper {
      */
     int startByRunId(String runId);
 
+    int stopJobTask(@Param("runId") String runId, @Param("nodeIds")List<Integer> nodeIds);
+
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java

@@ -148,6 +148,14 @@ public interface RunRecordingMapper {
     int stopByRunId(String runId);
 
     /**
+     * 停止 基于runid和flowId
+     * @param runId
+     * @param flowId
+     * @return
+     */
+    int stopByRunIdAndFlowId(@Param("runId") String runId, @Param("flowId")Integer flowId);
+
+    /**
      * 根据runid开始工作流调度
      * @param runId
      * @return

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

@@ -59,4 +59,11 @@ public interface JobTaskService {
     void deleteInIds(List<Integer> ids);
 
     int removeByRunId(String runId);
+
+    /**
+     * 暂停任务节点
+     * @param runId 运行标识
+     * @param nodeIds 要暂停的节点信息
+     */
+    void stopJobTask(String runId,List<Integer> nodeIds);
 }

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

@@ -94,4 +94,9 @@ public class JobTaskServiceImpl implements JobTaskService {
     public int removeByRunId(String runId) {
         return jobTaskMapper.deleteByRunId(runId);
     }
+
+    @Override
+    public void stopJobTask(String runId, List<Integer> nodeIds) {
+        jobTaskMapper.stopJobTask(runId, nodeIds);
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/util/FlowNodeRelyParseUtil.java

@@ -23,6 +23,7 @@ public class FlowNodeRelyParseUtil {
      * @param nodeAllList          所有的节点信息
      * @param startNodeName        开始的节点名称
      * @param allNodeDependencyKey 所有的主键信息
+     * @param hasStart 是否包含选中节点
      * @return 返回解析之后的节点
      */
     public static Set<NodeRelyDto> parseThisVersionNodeRely(List<Node> nodeAllList, String startNodeName, List<NodeDependencyKey> allNodeDependencyKey, boolean hasStart) {

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskMapper.xml

@@ -460,7 +460,12 @@
 
   <update id="stopByRunId"  parameterType="string">
     update job_task set trigger_status = '0'
-    where run_id = #{runId,jdbcType=INTEGER}
+    where run_id = #{runId,jdbcType=VARCHAR}
+  </update>
+
+  <update id="stopByRunIdAndFlowId" parameterType="string">
+    update job_task set trigger_status = '0'
+    where run_id = #{runId,jdbcType=VARCHAR} and flow_id = #{flowId,jdbcType=INTEGER}
   </update>
 
   <update id="startByRunId" parameterType="string">
@@ -495,4 +500,12 @@
     where run_id = #{runId,jdbcType=VARCHAR}
   </delete>
 
+  <update id="stopJobTask">
+    update job_task set trigger_status = '0'
+      where run_id = #{runId,jdbcType=VARCHAR} and id in
+      <foreach collection="nodeIds" item="nodeId" separator="," open="(" close=")">
+        #{nodeId,jdbcType=INTEGER}
+      </foreach>
+  </update>
+
 </mapper>
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml

@@ -683,6 +683,12 @@
             and flow_status != '4'
   </update>
 
+  <update id="stopByRunIdAndFlowId" parameterType="string">
+    update run_recording set flow_status = '3', stop_count = (stop_count+1)
+     where run_id = #{runId,jdbcType=VARCHAR} and flow_id = #{flowId, jdbcType=INTEGER}
+            and flow_status != '4'
+  </update>
+
   <update id="startByRunId" parameterType="string">
      update run_recording set flow_status = '2', stop_count = 0
      where run_id = #{runId,jdbcType=VARCHAR}

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/expand/ExtendedConfiguration.java

 package com.byit.dto.expand;
 
+import java.io.Serializable;
+
 /**
  * 扩展配置
  *
  * @author huangfu
  * @date 2020年10月20日15:11:12
  */
-public class ExtendedConfiguration {
+public class ExtendedConfiguration implements Serializable {
+    private static final long serialVersionUID = -3666177188893915790L;
 }

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/operating/StopNodeParam.java

+package com.byit.dto.operating;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 节点暂停参数
+ *
+ * @author huangfu
+ */
+@Data
+public class StopNodeParam implements Serializable {
+    private static final long serialVersionUID = 7180469480468848733L;
+    /**
+     * 工作流名称
+     */
+    private String flowName;
+    /**
+     * 工作空间名称
+     */
+    private String workspaceName;
+    /**
+     * 节点名称
+     */
+    private String nodeName;
+}