Commit aa66c071 by guominglei

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

parent a1541784
...@@ -8,6 +8,7 @@ import org.springframework.web.bind.annotation.RequestMapping; ...@@ -8,6 +8,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.text.ParseException;
/** /**
* @description: 工作流操作的API接口 * @description: 工作流操作的API接口
...@@ -38,14 +39,14 @@ public class ApiFlowController { ...@@ -38,14 +39,14 @@ public class ApiFlowController {
@PostMapping("start") @PostMapping("start")
@ApiOperation("开始工作流的调度,将工作流启用调度") @ApiOperation("开始工作流的调度,将工作流启用调度")
public String start(String flowName, String workspaceName){ public String start(String flowName, String workspaceName) throws ParseException {
apiFlowService.start(flowName, workspaceName); apiFlowService.start(flowName, workspaceName);
return "SUCCESS"; return "SUCCESS";
} }
@PostMapping("repealSchedule") @PostMapping("repealSchedule")
@ApiOperation("撤销工作流调度,") @ApiOperation("撤销工作流调度,")
public String repealSchedule(String flowName, String workspaceName){ public String repealSchedule(String flowName, String workspaceName) throws ParseException {
apiFlowService.repealSchedule(flowName, workspaceName); apiFlowService.repealSchedule(flowName, workspaceName);
return "SUCCESS"; return "SUCCESS";
} }
......
...@@ -2,6 +2,8 @@ package com.byit.service; ...@@ -2,6 +2,8 @@ package com.byit.service;
import com.byit.job.dto.plugin.PluginFlow; import com.byit.job.dto.plugin.PluginFlow;
import java.text.ParseException;
/** /**
* @description: 工作流的api请求业务处理接口 * @description: 工作流的api请求业务处理接口
* @author: gml * @author: gml
...@@ -11,9 +13,9 @@ public interface ApiFlowService { ...@@ -11,9 +13,9 @@ public interface ApiFlowService {
void deleteFlow(String flowName, String workspaceName); void deleteFlow(String flowName, String workspaceName);
void start(String flowName, String workspaceName); void start(String flowName, String workspaceName) throws ParseException;
void repealSchedule(String flowName, String workspaceName); void repealSchedule(String flowName, String workspaceName) throws ParseException;
void killSchedule(String flowName, String workspaceName); void killSchedule(String flowName, String workspaceName);
......
...@@ -22,6 +22,7 @@ import org.springframework.stereotype.Service; ...@@ -22,6 +22,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.text.ParseException;
import java.util.*; import java.util.*;
/** /**
...@@ -468,15 +469,45 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -468,15 +469,45 @@ public class ApiFlowServiceImpl implements ApiFlowService {
jobTaskMapper.deleteByRunId(runRecording.getRunId()); jobTaskMapper.deleteByRunId(runRecording.getRunId());
}); });
//删除工作流及下属节点 //删除工作流及下属节点
deleteFlowAndNode(flow);
} }
@Override /**
public void start(String flowName, String workspaceName) { * 删除工作流及下属节点
* @param flow
*/
private void deleteFlowAndNode(Flow flow) {
List<Node> nodeList = nodeMapper.findNodeByFlowId(flow.getFlowId());
if (null != nodeList && !nodeList.isEmpty()){
//删除节点的依赖
nodeList.forEach(node -> {
nodeDependencyMapper.deleteByNodeId(node.getNodeId());
if (StringUtils.isNotEmpty(node.getIsVirtual()) && NodePropertyEnum.IS_VIRTUAL.getCode().equals(node.getIsVirtual())){
Flow innerFlow = new Flow();
innerFlow.setFlowId(node.getMapFlowId());
deleteFlowAndNode(innerFlow);
}
});
//删除所有的节点
nodeMapper.deleteByFlowId(flow.getFlowId());
}
}
@Override
public void start(String flowName, String workspaceName) throws ParseException {
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()), "内嵌工作流不允许开启调度!");
//撤销工作流调度
updateFlowStart(flow, true);
log.info("工作流【{}】调度开始成功!", flowName);
} }
@Override @Override
public void repealSchedule(String flowName, String workspaceName) { public void repealSchedule(String flowName, String workspaceName) throws ParseException {
log.info("撤销工作空间【{}】---工作流【{}】调度", workspaceName, flowName); log.info("撤销工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
Workspace workspace = workspaceMapper.getByName(workspaceName); Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在"); ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
...@@ -494,9 +525,34 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -494,9 +525,34 @@ public class ApiFlowServiceImpl implements ApiFlowService {
jobTaskMapper.deleteByRunId(runRecording.getRunId()); jobTaskMapper.deleteByRunId(runRecording.getRunId());
}); });
//撤销工作流调度
updateFlowStart(flow, false);
log.info("工作流【{}】调度撤销成功!", flowName);
}
/**
* 更改工作流的启动状态
* @param flow 工作流
* @param isStart 是否启动
*/
private void updateFlowStart(Flow flow, boolean isStart) throws ParseException {
if (isStart){
flow.setStartUp(FlowPropertyEnum.IS_START.getCode());
//如果是周期调度修改下次执行时间
if (FlowPropertyEnum.SCHEDULE_MODE.getCode().equals(flow.getExecType())){
flow.setTriggerNextTime(new CronExpression(flow.getFlowCron()).getNextValidTimeAfter(new Date()).getTime());
}
}else {
flow.setStartUp(FlowPropertyEnum.NO_START.getCode()); flow.setStartUp(FlowPropertyEnum.NO_START.getCode());
}
flowMapper.updateByIdSelective(flow); flowMapper.updateByIdSelective(flow);
log.info("工作流【{}】调度撤销成功!", flowName); List<Node> nodeList = nodeMapper.findVirtualByFlowId(flow.getFlowId());
if (null != nodeList && nodeList.size() > 0 ){
for (Node node : nodeList){
Flow innerFlow = flowMapper.getById(node.getMapFlowId());
updateFlowStart(innerFlow, isStart);
}
}
} }
@Override @Override
......
...@@ -55,4 +55,17 @@ public interface NodeMapper { ...@@ -55,4 +55,17 @@ public interface NodeMapper {
* @return * @return
*/ */
Node getByNameAndFlow(@Param("nodeName") String nodeName, @Param("flowId") Integer flowId); Node getByNameAndFlow(@Param("nodeName") String nodeName, @Param("flowId") Integer flowId);
/**
* 查询工作流下的所有节点
* @param flowId
* @return
*/
List<Node> findNodeByFlowId(Integer flowId);
/**
* 删除工作流下的所有节点
* @param flowId
*/
int deleteByFlowId(Integer flowId);
} }
\ No newline at end of file
...@@ -91,6 +91,15 @@ ...@@ -91,6 +91,15 @@
and node_name = #{nodeName,jdbcType=VARCHAR} and node_name = #{nodeName,jdbcType=VARCHAR}
</select> </select>
<select id="findNodeByFlowId" parameterType="integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from node
where flow_id = #{flowId,jdbcType=INTEGER}
</select>
<delete id="deleteVirtualNode"> <delete id="deleteVirtualNode">
delete from node delete from node
where flow_id = #{flowId,jdbcType=INTEGER} where flow_id = #{flowId,jdbcType=INTEGER}
...@@ -103,6 +112,11 @@ ...@@ -103,6 +112,11 @@
where node_id = #{nodeId,jdbcType=INTEGER} where node_id = #{nodeId,jdbcType=INTEGER}
</delete> </delete>
<delete id="deleteByFlowId" parameterType="java.lang.Integer">
delete from node
where flow_id = #{flowId}
</delete>
<insert id="insertSelective" useGeneratedKeys="true" keyProperty="nodeId" parameterType="com.byit.model.Node"> <insert id="insertSelective" useGeneratedKeys="true" keyProperty="nodeId" parameterType="com.byit.model.Node">
<!-- generated @mbg.generated date: 2019-12-31 --> <!-- generated @mbg.generated date: 2019-12-31 -->
insert into node insert into node
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment