Commit bf8f12ff by huangfusuper

Merge remote-tracking branch 'origin/developer' into developer

parents 776a915e 6fbd5c6f
...@@ -503,11 +503,6 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -503,11 +503,6 @@ public class ApiFlowServiceImpl implements ApiFlowService {
waitingTask.setScheduleType(ScheduleTypeEnum.REPEAT.getCode()); waitingTask.setScheduleType(ScheduleTypeEnum.REPEAT.getCode());
waitingTask.setOperator(userName); waitingTask.setOperator(userName);
//查询当前节点的依赖节点
// if (dependNodeList != null && dependNodeList.size() > 0){
// waitingTask.setNodeDepend(Joiner.on(",").join(dependNodeList));
// }
List<WaitingTask> waitingTaskList = new ArrayList<>(); List<WaitingTask> waitingTaskList = new ArrayList<>();
waitingTaskList.add(waitingTask); waitingTaskList.add(waitingTask);
...@@ -517,9 +512,26 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -517,9 +512,26 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//查询依赖本节点的节点,并添加到集合中 //查询依赖本节点的节点,并添加到集合中
List<Integer> subNodeIdList = nodeDependencyMapper.findSubNodeList(node.getNodeId()); List<Integer> subNodeIdList = nodeDependencyMapper.findSubNodeList(node.getNodeId());
Set<Integer> nodeIdSet = new HashSet<>(); Set<Integer> nodeIdSet = new HashSet<>();
Set<Integer> runNodeIdSet = new HashSet<>();
runNodeIdSet.add(node.getNodeId());
nodeIdSet.add(node.getNodeId()); nodeIdSet.add(node.getNodeId());
//获取全部的需要重跑的节点
if (subNodeIdList != null && subNodeIdList.size() > 0){
Queue<Integer> queue = new LinkedList<>();
subNodeIdList.forEach(nodeId -> {
queue.offer(nodeId);
runNodeIdSet.add(nodeId);
});
while(!queue.isEmpty()){
subNodeIdList = nodeDependencyMapper.findSubNodeList(queue.poll());
if (subNodeIdList != null && subNodeIdList.size() > 0){ if (subNodeIdList != null && subNodeIdList.size() > 0){
addDependNode(runInfo.getRunId(), triggerTime, waitingTaskList, subNodeIdList, userName, nodeIdSet); subNodeIdList.forEach(nodeId->{
queue.offer(nodeId);
runNodeIdSet.add(nodeId);
});
}
}
addDependNode(runInfo.getRunId(), triggerTime, waitingTaskList, subNodeIdList, userName, nodeIdSet, runNodeIdSet);
} }
} }
...@@ -627,7 +639,7 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -627,7 +639,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
} }
} }
private void addDependNode(String runId, Long triggerTime, List<WaitingTask> waitingTaskList, List<Integer> subNodeIdList, String userName, Set<Integer> nodeIdSet) { private void addDependNode(String runId, Long triggerTime, List<WaitingTask> waitingTaskList, List<Integer> subNodeIdList, String userName, Set<Integer> nodeIdSet, Set<Integer> runNodeIdSet) {
subNodeIdList.forEach(childNodeId -> { subNodeIdList.forEach(childNodeId -> {
JobTaskRunLog jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndNodeId(runId, childNodeId); JobTaskRunLog jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndNodeId(runId, childNodeId);
WaitingTask waitingTask = new WaitingTask(); WaitingTask waitingTask = new WaitingTask();
...@@ -641,7 +653,13 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -641,7 +653,13 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//查询当前节点的依赖节点 //查询当前节点的依赖节点
List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(jobTaskRunLog.getNodeId()); List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(jobTaskRunLog.getNodeId());
if (dependNodeIdList != null && dependNodeIdList.size() > 0){ if (dependNodeIdList != null && dependNodeIdList.size() > 0){
waitingTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList)); List<Integer> realDependNodeIdList = new ArrayList<>();
for (Integer dependNodeId : dependNodeIdList) {
if (!runNodeIdSet.add(dependNodeId)){
realDependNodeIdList.add(dependNodeId);
}
}
waitingTask.setNodeDepend(Joiner.on(",").join(realDependNodeIdList));
} }
if (nodeIdSet.add(childNodeId)){ if (nodeIdSet.add(childNodeId)){
waitingTaskList.add(waitingTask); waitingTaskList.add(waitingTask);
...@@ -649,7 +667,7 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -649,7 +667,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//查询依赖于当前节点的下级节点 //查询依赖于当前节点的下级节点
List<Integer> childNodeIdList = nodeDependencyMapper.findSubNodeList(childNodeId); List<Integer> childNodeIdList = nodeDependencyMapper.findSubNodeList(childNodeId);
if (childNodeIdList != null && childNodeIdList.size() > 0){ if (childNodeIdList != null && childNodeIdList.size() > 0){
addDependNode(runId, triggerTime, waitingTaskList, childNodeIdList, userName, nodeIdSet); addDependNode(runId, triggerTime, waitingTaskList, childNodeIdList, userName, nodeIdSet, runNodeIdSet);
} }
}); });
} }
......
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