Commit 7461ec3f by huangfusuper

修正重新发布节点时,删除节点不生效的BUG

parent c303fc0a
package com.byit.service;
/**
* 工作流节点校验
* @author huangfu
*/
public interface ApiNodeCheckService {
/**
* 工作流校验
* @param mainFlowId 主工作流的id
*/
void checkFlowNode(Integer mainFlowId);
}
......@@ -20,6 +20,7 @@ import com.byit.model.RunRecording;
import com.byit.model.*;
import com.byit.model.vo.RunRecordingVo;
import com.byit.service.ApiFlowService;
import com.byit.service.ApiNodeCheckService;
import com.byit.util.ApiFlowDagCheck;
import com.byit.util.IDGenerationStrategy;
import com.byit.util.lock.RedissLockUtil;
......@@ -61,6 +62,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
private FlowMapper flowMapper;
@Resource
private ApiNodeCheckService apiNodeCheckService;
@Resource
private FlowVersionMapper flowVersionMapper;
@Resource
......@@ -128,6 +132,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//生成工作流版本
log.debug("生成版本");
FlowVersion flowVersion = saveFlowVersion(flow);
apiNodeCheckService.checkFlowNode(flow.getFlowId());
log.debug("插件端通过API保存成功!");
}
......@@ -220,6 +226,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
HashMap<String, Integer> nameIdRel = new HashMap<>();
//保存节点信息
for(PluginBaseNode pluginNode : pluginNodeList){
//查询这个节点是否存在 存在就修改节点 不存在就新建节点
Node node = nodeMapper.getByNameAndFlow(pluginNode.getName(), flow.getFlowId());
if (null == node){
node = new Node();
......
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 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
*/
@Service
@Transactional(rollbackFor = Exception.class)
@Slf4j
public class ApiNodeCheckServiceImpl implements ApiNodeCheckService {
private final NodeService nodeService;
private final FlowService flowService;
public ApiNodeCheckServiceImpl(NodeService nodeService, FlowService flowService) {
this.nodeService = nodeService;
this.flowService = flowService;
}
@Override
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);
}
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);
}
}
}
}
......@@ -95,4 +95,6 @@ public interface NodeMapper {
List<Node> findByFlowIdAndName(@Param("flowId") Integer flowId, @Param("nodeNameList")List<String> nodeNameList);
List<Node> findbyFlowIds(@Param("flowIdList")List<Integer> flowIdList);
int deleteNodeByIds(@Param("ids") List<Integer> ids);
}
\ No newline at end of file
......@@ -26,6 +26,6 @@ public interface NodeService {
Node findNodeById(Integer id);
int deleteNode(List<Integer> ids);
}
......@@ -30,4 +30,10 @@ public class NodeServiceImpl implements NodeService {
public Node findNodeById(Integer id) {
return nodeMapper.getById(id);
}
@Override
public int deleteNode(List<Integer> ids) {
return nodeMapper.deleteNodeByIds(ids);
}
}
......@@ -49,13 +49,6 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
@Override
public void logAndRunRecordingFailFast(RunRecording runRecording) {
String runId = runRecording.getRunId();
//Integer flowId = runRecording.getFlowId();
//查询该实例对应的所由日志节点
//List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
//List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
//List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(flowId);
//筛选没在日志里面的节点
//List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
//将这些节点置为失败并将实例也置为失败
//保存错误日志节点
//第一将非虚节点的节点保存进日志,同时保存虚节点,同时筛选出end节点 不作处理,让他自己执行队形的end节点(为了执行结束事件)
......@@ -63,6 +56,7 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
//查询task所有的对应节点 这里需要判断不是虚节点 在进行日志插入之后需要将对应的节点删除
List<JobTask> byRunId = jobTaskService.findByRunId(runId);
byRunId.forEach(task ->{
//保存剩余节点的日志信息
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
BeanUtils.copyProperties(task,log);
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
......@@ -76,10 +70,11 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
log.setFlowName(runRecording.getFlowName());
log.setTriggerTime(thisTime);
jobTaskRunLogService.saveJobTaskRunLog(log);
//筛选虚节点
if ("0".equals(task.getIsVirtual())) {
virtualTasks.add(task);
}
//删除这些节点
jobTaskService.removeMythJobTaskById(task.getId());
});
//执行实例的快速失败
......@@ -91,6 +86,7 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
runRecording.setFlowRunResult(RunRecordingEnum.RUN_FLOW_FAILURE.getCode());
runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
runRecordingService.updateRunRecordingById(runRecording);
//执行工作流的完成事件
applicationEventPublisher.publishEvent(new EndFlowEvent(this,runRecording.getFlowId()));
virtualTasks.forEach(virtualTask -> {
......
......@@ -90,13 +90,14 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
jobTaskRunLog.setTriggerTime(thisDate);
jobTaskRunLog.setStartTime(thisDate);
jobTaskRunLog.setEndTime(thisDate);
if("end".equals(jobTask.getNodeName())){
//未完成告警
// if("end".equals(jobTask.getNodeName())){
// //未完成告警
// jobTaskRunLog.setAlertEnd("0");
// }else{
// //已完成告警
// jobTaskRunLog.setAlertEnd("1");
// }
jobTaskRunLog.setAlertEnd("0");
}else{
//已完成告警
jobTaskRunLog.setAlertEnd("1");
}
//添加日志节点
jobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
......@@ -124,7 +125,8 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
.principal(virFlow.getPrincipal())
.startTime(new Date())
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(mainRecording.getIsAlarm())
//.isAlarm(mainRecording.getIsAlarm())
.isAlarm("1")
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_YES.getCode())
.workspaceId(virFlow.getWorkspaceId())
......
......@@ -149,6 +149,16 @@
)
</select>
<delete id="deleteNodeByIds">
delete from node
where node_id in(
<foreach collection="ids" item="id" separator=",">
#{id}
</foreach>
)
</delete>
<delete id="deleteVirtualNode">
delete from node
where flow_id = #{flowId,jdbcType=INTEGER}
......
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