Skip to content

B
byit-myth-job
Commit c602e122 by guominglei
Options

撤销工作流调度

parent 1abd5f2a
Showing with 107 additions and 18 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java
......@@ -34,7 +34,7 @@ public class ApiFlowController {
@PostMapping("delete")
@ApiOperation("删除工作流,若当前工作流被依赖则删除失败，只允许删除不被依赖的工作流，当前工作流依赖其他工作流不影响")
public String deleteFlow(String flowName, String workspaceName){
apiFlowService.deleteFlow(flowName);
apiFlowService.deleteFlow(flowName, workspaceName);
return "SUCCESS";
}
......
byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java
......@@ -10,7 +10,7 @@ import com.byit.job.dto.plugin.PluginPackage;
*/
public interface ApiFlowService {
void deleteFlow(String flowName);
void deleteFlow(String flowName, String workspaceName);
void start(String flowName, String workspaceName);
......
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -54,16 +54,25 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Resource
private NodeVersionDependencyMapper nodeVersionDependencyMapper;
@Resource
private RunRecordingMapper runRecordingMapper;
@Resource
private JobTaskMapper jobTaskMapper;
@Override
@Transactional(rollbackFor = Exception.class)
public void publishFlow(PluginPackage pluginPackage) throws Exception {
//校验参数
log.info("校验参数是否符合规范");
Workspace workspace = validate(pluginPackage);
//保存工作流
log.info("保存工作流和节点信息");
Flow flow = saveFlow(pluginPackage.getFlow(), workspace.getWorkspaceId(), false);
//生成工作流版本
log.info("生成版本");
FlowVersion flowVersion = saveFlowVersion(flow);
log.info("插件端通过API保存成功！");
}
/**
......@@ -434,8 +443,24 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
@Override
public void deleteFlow(String flowName) {
public void deleteFlow(String flowName, String workspaceName) {
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 , "工作流已在调度中不允许撤销调度！");
List<RunRecording> unStartRecordingList = runRecordingMapper.findUnStartByFlowId(flow.getFlowId());
//删除对应的task记录
unStartRecordingList.forEach(runRecording -> {
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) > 10000 , "工作流已在调度中不允许删除");
runRecordingMapper.deleteByRunId(runRecording.getRunId());
jobTaskMapper.deleteByRunId(runRecording.getRunId());
});
//删除工作流及下属节点
}
@Override
......@@ -445,7 +470,26 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Override
public void repealSchedule(String flowName, String workspaceName) {
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 , "工作流已在调度中不允许撤销调度！");
List<RunRecording> unStartRecordingList = runRecordingMapper.findUnStartByFlowId(flow.getFlowId());
//删除对应的task记录
unStartRecordingList.forEach(runRecording -> {
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) > 10000 , "工作流已在调度中不允许撤销调度");
runRecordingMapper.deleteByRunId(runRecording.getRunId());
jobTaskMapper.deleteByRunId(runRecording.getRunId());
});
flow.setStartUp(FlowPropertyEnum.NO_START.getCode());
flowMapper.updateByIdSelective(flow);
log.info("工作流【{}】调度撤销成功！", flowName);
}
@Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java
......@@ -51,4 +51,10 @@ public interface JobTaskMapper {
* @param jobTasks
*/
void deleteInId(@Param("jobTasks") List<JobTask> jobTasks);
/**
* 根据runid删除运行信息
* @param runId
*/
int deleteByRunId(String runId);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
......@@ -4,6 +4,8 @@ import com.byit.model.RunRecording;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* 运行记录表
* @author huangfu
......@@ -50,4 +52,22 @@ public interface RunRecordingMapper {
*/
int deleteById(Integer recordingId);
/**
* 根据工作流id查询正在调度中的工作流
* @param flowId
*/
List<RunRecording> findOnScheduleByFlowId(Integer flowId);
/**
* 根据工作流id查询未开始的工作流
* @param flowId
* @return
*/
List<RunRecording> findUnStartByFlowId(Integer flowId);
/**
* 根据runid删除工作流的运行记录
* @param runId
*/
int deleteByRunId(String runId);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/Generator-config.xml
......@@ -55,19 +55,7 @@
type="XMLMAPPER">
<property name="enableSubPackages" value="false"/>
</javaClientGenerator>
<table tableName="email_alarm" domainObjectName="EmailAlarm" />
<table tableName="flow" domainObjectName="Flow" />
<table tableName="flow_version" domainObjectName="FlowVersion" />
<table tableName="job_task" domainObjectName="JobTask" />
<table tableName="job_task_run_log" domainObjectName="JobTaskRunLog" />
<table tableName="job_task_schedule" domainObjectName="JobTaskSchedule" />
<table tableName="node" domainObjectName="Node" />
<table tableName="node_version" domainObjectName="NodeVersion" />
<table tableName="run_recording" domainObjectName="RunRecording" />
<table tableName="source_history" domainObjectName="SourceHistory" />
<table tableName="flow_dependent" domainObjectName="FlowDependent" />
<table tableName="node_dependency" domainObjectName="NodeDependency" />
<table tableName="workspace" domainObjectName="Workspace" />
<table tableName="node_version_dependency" domainObjectName="NodeVersionDependency" />
</context>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskMapper.xml
......@@ -344,4 +344,9 @@
delete from job_task
where id = #{id,jdbcType=INTEGER}
</delete>
<delete id="deleteByRunId" parameterType="java.lang.Integer">
delete from job_task
where run_id = #{runId,jdbcType=INTEGER}
</delete>
</mapper>
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -41,11 +41,34 @@
from run_recording
where recording_id = #{recordingId,jdbcType=INTEGER}
</select>
<select id="findOnScheduleByFlowId" parameterType="java.lang.Integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_id = #{flowId,jdbcType=INTEGER}
and flow_status not in ('1','4')
</select>
<select id="findUnStartByFlowId" parameterType="java.lang.Integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_id = #{flowId,jdbcType=INTEGER}
and flow_status = '1'
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2019-12-25 -->
delete from run_recording
where recording_id = #{recordingId,jdbcType=INTEGER}
</delete>
<delete id="deleteByRunId" parameterType="java.lang.Integer">
delete from run_recording
where run_id = #{runId,jdbcType=INTEGER}
</delete>
<insert id="saveRunRecording" parameterType="com.byit.model.RunRecording">
<!-- generated @mbg.generated date: 2019-12-25 -->
insert into run_recording
......
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