插件端开启调度和撤销调度

byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java

@@ -45,7 +45,7 @@ public class ApiFlowController {
     }
 
     @PostMapping("repealSchedule")
-    @ApiOperation("撤销工作流调度，")
+    @ApiOperation("撤销工作流调度，只可以撤销总工作流，若为内嵌工作流不允许撤销")
     public String repealSchedule(String flowName, String workspaceName) throws ParseException {
         apiFlowService.repealSchedule(flowName, workspaceName);
         return "SUCCESS";
@@ -66,8 +66,8 @@ public class ApiFlowController {
 
     @PostMapping("reStartSchedule")
     @ApiOperation("重新开始某次调度")
-    public String reStartSchedule(String runId){
-        apiFlowService.reStartSchedule(runId);
+    public String reStartSchedule(String runIds){
+        apiFlowService.reStartSchedule(runIds);
         return "SUCCESS";
     }
 

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

@@ -21,7 +21,7 @@ public interface ApiFlowService {
 
     String stopSchedule(String flowName, String workspaceName);
 
-    void reStartSchedule(String runId);
+    void reStartSchedule(String runIds);
 
     void publishFlow(String param) throws Exception;
 

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

@@ -340,7 +340,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
                 checkFlow(workspace, (PluginFlow) node, flowNameSet, fromFlowSet, toFlowSet, lineMap);
             }else {
                 ValidationUtil.dataNotBank(((PluginNode) node).getJobType(), "节点类型不允许为空！");
-
+                if ("start".equals(node.getName()) || "end".equals(node.getName())){
+                    ((PluginNode) node).getConfig().setNodeCron(flow.getConfig().getFlowCron());
+                }
                 if(FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())) {
                     ValidationUtil.dataNotNull(((PluginNode) node).getConfig(), "节点配置信息不允许为空！");
                     ValidationUtil.dataNotBank(((PluginNode) node).getConfig().getNodeCron(), "工作流设置为不跟随调度时节点必须设置调度时间！");
@@ -562,12 +564,35 @@ public class ApiFlowServiceImpl implements ApiFlowService {
 
     @Override
     public String stopSchedule(String flowName, String workspaceName) {
-        return null;
+        log.info("停止工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
+        Workspace workspace = workspaceMapper.getByName(workspaceName);
+        ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
+        Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
+        ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
+        ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许停止调度！");
+        //判断是否在调度中
+        List<RunRecording> recordingList = runRecordingMapper.findOnScheduleByFlowId(flow.getFlowId());
+        ValidationUtil.isTrueValidation(null == recordingList || recordingList.size() == 0 , flowName + "工作流没有正在运行的调度！");
+        StringBuffer runids = new StringBuffer();
+        //暂停工作流调度
+        recordingList.forEach(runRecording -> {
+            runRecordingMapper.stopByRunId(runRecording.getRunId());
+            jobTaskMapper.stopByRunId(runRecording.getRunId());
+            runids.append(runRecording.getRunId() + ",");
+        });
+        return runids.toString();
     }
 
     @Override
-    public void reStartSchedule(String runId) {
-
+    public void reStartSchedule(String runIds) {
+        ValidationUtil.dataNotBank(runIds, "运行id不允许为空！");
+        List<String> runIdList = Arrays.asList(runIds.split(","));
+        runIdList.forEach(runId -> {
+            List<RunRecording>  runRecordList = runRecordingMapper.findStopRunCordByRunId(runId);
+            ValidationUtil.dataNotNull(runRecordList, "运行记录" + runId + "不存在暂停中运行记录！");
+            runRecordingMapper.startByRunId(runId);
+            jobTaskMapper.startByRunId(runId);
+        });
     }
 
 }

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

@@ -57,4 +57,17 @@ public interface JobTaskMapper {
      * @param runId
      */
     int deleteByRunId(String runId);
+
+    /**
+     * 根据runid暂停调度
+     * @param runId
+     * @return
+     */
+    int stopByRunId(String runId);
+    /**
+     * 根据runid开始调度
+     * @param runId
+     * @return
+     */
+    int startByRunId(String runId);
 }
\ No newline at end of file

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

@@ -75,4 +75,25 @@ public interface RunRecordingMapper {
      * @param runId
      */
     int deleteByRunId(String runId);
+
+    /**
+     * 根据runid暂停工作流调度
+     * @param runId
+     */
+    int stopByRunId(String runId);
+
+    /**
+     * 根据runid开始工作流调度
+     * @param runId
+     * @return
+     */
+    int startByRunId(String runId);
+
+    /**
+     * 根据runid查找停止的运行记录
+     * @param runId
+     * @return
+     */
+    List<RunRecording> findStopRunCordByRunId(String runId);
+
 }
\ No newline at end of file

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

@@ -331,6 +331,16 @@
     where id = #{id,jdbcType=INTEGER}
   </update>
 
+  <update id="stopByRunId"  parameterType="string">
+    update job_task set trigger_status = '0'
+    where run_id = #{runId,jdbcType=INTEGER}
+  </update>
+
+  <update id="startByRunId" parameterType="string">
+    update job_task set trigger_status = '1'
+    where run_id = #{runId,jdbcType=INTEGER}
+  </update>
+
   <!--多个删除-->
   <delete id="deleteInId" parameterType="com.byit.model.JobTask">
     delete from job_task

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

@@ -66,6 +66,14 @@
         and flow_status = '1'
   </select>
 
+  <select id="findStopRunCordByRunId" parameterType="string" resultMap="BaseResultMap">
+    select
+    <include refid="Base_Column_List" />
+    from run_recording
+    where run_id = #{runId,jdbcType=VARCHAR}
+        and flow_status = '3'
+  </select>
+
   <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2019-12-25 -->
     delete from run_recording
@@ -320,4 +328,25 @@
     </set>
     where flow_id = #{flowId,jdbcType=INTEGER} and run_id = #{runId,jdbcType=VARCHAR}
   </update>
+
+  <update id="stopByRunId" parameterType="string">
+     update run_recording set flow_status = '3'
+     where recording_id in(
+        select recording_id
+        from run_recording
+        where run_id = #{runId,jdbcType=VARCHAR}
+            and flow_status in ('1', '2')
+     )
+  </update>
+
+  <update id="startByRunId" parameterType="string">
+     update run_recording set flow_status = '2'
+     where recording_id in(
+        select recording_id
+        from run_recording
+        where run_id = #{runId,jdbcType=VARCHAR}
+            and flow_status = '3'
+     )
+  </update>
+
 </mapper>
\ No newline at end of file