Spider/spider-flow
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复多个循环的聚合变量BUG

Browse Source
This commit is contained in:
mxd 2019-10-27 20:56:54 +08:00
parent 42b1b7e769
commit 89c0bf0416
3 changed files with 27 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
spider-flow-core/src/main/java/org/spiderflow/core/Spider.java
View File

@ -165,6 +165,7 @@ public class Spider {
String loopVariableName = node.getStringJsonValue(ShapeExecutor.LOOP_VARIABLE_NAME);
if(executor instanceof LoopExecutor){
variables.put(LoopExecutor.BEFORE_LOOP_VARIABLE, variables);
variables.put(LoopExecutor.LOOP_NODE_KEY + node.getNodeId(), new CountDownLatch(loopCount));
variables.put(LoopJoinExecutor.VARIABLE_CONTEXT + node.getNodeId(), new LinkedBlockingQueue<>(loopCount));
}

2
spider-flow-core/src/main/java/org/spiderflow/core/executor/shape/LoopExecutor.java
View File

@ -17,6 +17,8 @@ public class LoopExecutor implements ShapeExecutor{
public static final String LOOP_NODE_KEY = "__loop_node_";
public static final String BEFORE_LOOP_VARIABLE = "__loop_before_variable";
@Override
public void execute(SpiderNode node, SpiderContext context, Map<String,Object> variables) {
}

43
spider-flow-core/src/main/java/org/spiderflow/core/executor/shape/LoopJoinExecutor.java
View File

@ -40,26 +40,31 @@ public class LoopJoinExecutor implements ShapeExecutor {
variableCollection.add(variables);
CountDownLatch countDownLatch = (CountDownLatch) variables.get(LoopExecutor.LOOP_NODE_KEY + joinNodeId);
if (countDownLatch != null) {
synchronized (countDownLatch) {
countDownLatch.countDown();
boolean isDone = countDownLatch.getCount() == 0L;
if(isDone){
//清空原有
variableCollection.stream()
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList())))
.forEach((k,v)->{
String key = "@" + k;
if(variables.containsKey(key) == false || k.startsWith("@")){
if(key.startsWith("@@")){
key = k;
}
variables.put(key, v);
countDownLatch.countDown();
boolean isDone = countDownLatch.getCount() == 0L;
if(isDone){
Map<String, Object> beforeLoopVariable = (Map<String, Object>) variables.get(LoopExecutor.BEFORE_LOOP_VARIABLE);
variableCollection.stream()
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList())))
.forEach((k,v)->{
String key = "@" + k;
if(variables.containsKey(key) == false || k.startsWith("@")){
if(key.startsWith("@@")){
key = k;
}
});
}
return isDone;
//清除掉原有变量
variables.remove(k);
variables.put(key, v);
}
});
//与循环前的变量进行合并
variables.putAll(beforeLoopVariable);
//删除掉多余出来的聚合变量
beforeLoopVariable.forEach((k,v)->{
variables.remove("@" + k);
});
return isDone;
}
} else {
context.error("找不到等待节点：{}" + node.getStringJsonValue(JOIN_NODE_ID));