feat: 增强工作流循环双输入运行语义
- 支持次数、数组及二者组合的循环执行计划 - 保持旧循环定义和上游数组读取兼容 - 补充解析、边界与前缀遍历测试
This commit is contained in:
@@ -35,6 +35,9 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 支持纯次数、数组遍历和两者组合限制的显式循环节点。
|
||||||
|
*/
|
||||||
public class LoopNode extends BaseNode {
|
public class LoopNode extends BaseNode {
|
||||||
private static final long serialVersionUID = 1L;
|
private static final long serialVersionUID = 1L;
|
||||||
|
|
||||||
@@ -43,16 +46,70 @@ public class LoopNode extends BaseNode {
|
|||||||
Integer.getInteger(
|
Integer.getInteger(
|
||||||
"tinyflow.loop.direct-index.max-items", 64));
|
"tinyflow.loop.direct-index.max-items", 64));
|
||||||
|
|
||||||
|
/** 旧版单循环参数。 */
|
||||||
private Parameter loopVar;
|
private Parameter loopVar;
|
||||||
|
/** 可选的循环次数参数。 */
|
||||||
|
private Parameter loopCount;
|
||||||
|
/** 可选的数组输入参数。 */
|
||||||
|
private Parameter loopItems;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 获取旧版循环参数。
|
||||||
|
*
|
||||||
|
* @return 旧版循环参数
|
||||||
|
*/
|
||||||
public Parameter getLoopVar() {
|
public Parameter getLoopVar() {
|
||||||
return loopVar;
|
return loopVar;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 设置旧版循环参数。
|
||||||
|
*
|
||||||
|
* @param loopVar 旧版循环参数
|
||||||
|
*/
|
||||||
public void setLoopVar(Parameter loopVar) {
|
public void setLoopVar(Parameter loopVar) {
|
||||||
this.loopVar = loopVar;
|
this.loopVar = loopVar;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 获取显式循环次数参数。
|
||||||
|
*
|
||||||
|
* @return 循环次数参数
|
||||||
|
*/
|
||||||
|
public Parameter getLoopCount() {
|
||||||
|
return loopCount;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 设置显式循环次数参数。
|
||||||
|
*
|
||||||
|
* @param loopCount 循环次数参数
|
||||||
|
*/
|
||||||
|
public void setLoopCount(Parameter loopCount) {
|
||||||
|
this.loopCount = loopCount;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 获取待遍历数组参数。
|
||||||
|
*
|
||||||
|
* @return 数组参数
|
||||||
|
*/
|
||||||
|
public Parameter getLoopItems() {
|
||||||
|
return loopItems;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 设置待遍历数组参数。
|
||||||
|
*
|
||||||
|
* @param loopItems 数组参数
|
||||||
|
*/
|
||||||
|
public void setLoopItems(Parameter loopItems) {
|
||||||
|
this.loopItems = loopItems;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@inheritDoc}
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public Map<String, Object> execute(Chain chain) {
|
public Map<String, Object> execute(Chain chain) {
|
||||||
MaterializationPlan[] planHolder = new MaterializationPlan[1];
|
MaterializationPlan[] planHolder = new MaterializationPlan[1];
|
||||||
@@ -164,9 +221,17 @@ public class LoopNode extends BaseNode {
|
|||||||
boolean directlyIndexed = false;
|
boolean directlyIndexed = false;
|
||||||
boolean numericLoop = false;
|
boolean numericLoop = false;
|
||||||
Object loopValue = null;
|
Object loopValue = null;
|
||||||
|
boolean usesExplicitInputs =
|
||||||
|
loopCount != null || loopItems != null;
|
||||||
|
Integer requestedCount = usesExplicitInputs
|
||||||
|
? resolveRequestedLoopCount(chain, loopContext)
|
||||||
|
: null;
|
||||||
if (loopContext.iterableInputStored) {
|
if (loopContext.iterableInputStored) {
|
||||||
// 已物化输入不再解析原始参数,避免大集合在每轮反序列化和路径解析。
|
// 已物化输入不再解析原始参数,避免大集合在每轮反序列化和路径解析。
|
||||||
shouldLoopCount = loopContext.iterableSize;
|
shouldLoopCount = planIterableIterations(
|
||||||
|
loopContext,
|
||||||
|
loopContext.iterableSize,
|
||||||
|
requestedCount);
|
||||||
storedIterable = true;
|
storedIterable = true;
|
||||||
} else {
|
} else {
|
||||||
LoopInputReference storedInput =
|
LoopInputReference storedInput =
|
||||||
@@ -178,17 +243,80 @@ public class LoopNode extends BaseNode {
|
|||||||
storedInput.getItemCount();
|
storedInput.getItemCount();
|
||||||
loopContext.iterableInputStored = true;
|
loopContext.iterableInputStored = true;
|
||||||
loopContext.inputExternalized = true;
|
loopContext.inputExternalized = true;
|
||||||
shouldLoopCount =
|
shouldLoopCount = planIterableIterations(
|
||||||
storedInput.getItemCount();
|
loopContext,
|
||||||
|
storedInput.getItemCount(),
|
||||||
|
requestedCount);
|
||||||
storedIterable = true;
|
storedIterable = true;
|
||||||
persistLoopStack(chain, loopStack);
|
persistLoopStack(chain, loopStack);
|
||||||
|
} else if (loopItems != null) {
|
||||||
|
loopValue = resolveParameterValue(
|
||||||
|
chain, loopItems, "loop items");
|
||||||
|
Iterable<?> iterableInput =
|
||||||
|
toIterableInput(loopValue);
|
||||||
|
if (iterableInput == null) {
|
||||||
|
throw invalidLoopItemsValue(loopValue);
|
||||||
|
}
|
||||||
|
int knownSize = knownInputSize(loopValue);
|
||||||
|
if (knownSize >= 0) {
|
||||||
|
shouldLoopCount = planIterableIterations(
|
||||||
|
loopContext,
|
||||||
|
knownSize,
|
||||||
|
requestedCount);
|
||||||
|
checkExplicitLoopIterations(
|
||||||
|
chain, shouldLoopCount, true);
|
||||||
} else {
|
} else {
|
||||||
Map<String, Object> loopVars =
|
shouldLoopCount = requestedCount == null
|
||||||
chain.getExecutionState().resolveParameters(
|
? -1
|
||||||
this,
|
: requestedCount;
|
||||||
Collections.singletonList(
|
}
|
||||||
loopVar));
|
|
||||||
loopValue = loopVars.get(loopVar.getName());
|
if (knownSize >= 0
|
||||||
|
&& shouldLoopCount <= DIRECT_INDEX_MAX_ITEMS) {
|
||||||
|
directlyIndexed = true;
|
||||||
|
} else {
|
||||||
|
/*
|
||||||
|
* 有次数限制时只物化所需前缀。前缀快照仅属于当前循环,
|
||||||
|
* 不替换上游完整数组,避免影响其他下游节点。
|
||||||
|
*/
|
||||||
|
Iterable<?> materializedInput =
|
||||||
|
requestedCount == null
|
||||||
|
? iterableInput
|
||||||
|
: limitIterable(
|
||||||
|
iterableInput,
|
||||||
|
requestedCount);
|
||||||
|
boolean externalizeInput =
|
||||||
|
requestedCount == null
|
||||||
|
|| (knownSize >= 0
|
||||||
|
&& requestedCount >= knownSize);
|
||||||
|
loopContext.resultId =
|
||||||
|
chain.getStateInstanceId() + ":" + UUID.randomUUID();
|
||||||
|
loopContext.materializingInput = true;
|
||||||
|
loopContext.materializationClaimId =
|
||||||
|
chain.currentFencingClaimId();
|
||||||
|
loopContext.materializationClaimGeneration =
|
||||||
|
chain.currentClaimGeneration();
|
||||||
|
persistLoopStack(chain, loopStack);
|
||||||
|
if (planHolder == null) {
|
||||||
|
throw new IllegalStateException(
|
||||||
|
"Loop materialization plan holder is unavailable");
|
||||||
|
}
|
||||||
|
planHolder[0] = new MaterializationPlan(
|
||||||
|
loopContext.resultId,
|
||||||
|
materializedInput,
|
||||||
|
loopItems,
|
||||||
|
externalizeInput,
|
||||||
|
loopContext.materializationClaimId,
|
||||||
|
loopContext.materializationClaimGeneration);
|
||||||
|
return waitingResult();
|
||||||
|
}
|
||||||
|
} else if (loopCount != null) {
|
||||||
|
shouldLoopCount = requestedCount;
|
||||||
|
numericLoop = true;
|
||||||
|
planNumericIterations(loopContext, shouldLoopCount);
|
||||||
|
} else if (loopVar != null) {
|
||||||
|
loopValue = resolveParameterValue(
|
||||||
|
chain, loopVar, "legacy loop value");
|
||||||
Iterable<?> iterableInput =
|
Iterable<?> iterableInput =
|
||||||
toIterableInput(loopValue);
|
toIterableInput(loopValue);
|
||||||
if (iterableInput != null) {
|
if (iterableInput != null) {
|
||||||
@@ -220,6 +348,8 @@ public class LoopNode extends BaseNode {
|
|||||||
planHolder[0] = new MaterializationPlan(
|
planHolder[0] = new MaterializationPlan(
|
||||||
loopContext.resultId,
|
loopContext.resultId,
|
||||||
iterableInput,
|
iterableInput,
|
||||||
|
loopVar,
|
||||||
|
true,
|
||||||
loopContext.materializationClaimId,
|
loopContext.materializationClaimId,
|
||||||
loopContext.materializationClaimGeneration);
|
loopContext.materializationClaimGeneration);
|
||||||
return waitingResult();
|
return waitingResult();
|
||||||
@@ -231,6 +361,9 @@ public class LoopNode extends BaseNode {
|
|||||||
} else {
|
} else {
|
||||||
throw invalidLoopValue(loopValue);
|
throw invalidLoopValue(loopValue);
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
|
throw new IllegalArgumentException(
|
||||||
|
"Loop node requires loop count or loop items");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
checkExplicitLoopIterations(chain, shouldLoopCount, !numericLoop);
|
checkExplicitLoopIterations(chain, shouldLoopCount, !numericLoop);
|
||||||
@@ -329,10 +462,16 @@ public class LoopNode extends BaseNode {
|
|||||||
"loop-materialization:" + plan.resultId);
|
"loop-materialization:" + plan.resultId);
|
||||||
}
|
}
|
||||||
context.iterableSize = iterableSize;
|
context.iterableSize = iterableSize;
|
||||||
|
context.plannedIterations = iterableSize;
|
||||||
|
context.plannedIterationsResolved = true;
|
||||||
context.iterableInputStored = true;
|
context.iterableInputStored = true;
|
||||||
context.materializingInput = false;
|
context.materializingInput = false;
|
||||||
context.inputExternalized = externalizeMaterializedInput(
|
context.inputExternalized = plan.externalizeInput
|
||||||
chain, plan.resultId, iterableSize);
|
&& externalizeMaterializedInput(
|
||||||
|
chain,
|
||||||
|
plan.sourceParameter,
|
||||||
|
plan.resultId,
|
||||||
|
iterableSize);
|
||||||
persistLoopStack(chain, loopStack);
|
persistLoopStack(chain, loopStack);
|
||||||
return executeLocked(chain, true, null);
|
return executeLocked(chain, true, null);
|
||||||
}
|
}
|
||||||
@@ -349,9 +488,16 @@ public class LoopNode extends BaseNode {
|
|||||||
* @return 已替换热状态值时为 {@code true}
|
* @return 已替换热状态值时为 {@code true}
|
||||||
*/
|
*/
|
||||||
private boolean externalizeMaterializedInput(
|
private boolean externalizeMaterializedInput(
|
||||||
Chain chain, String resultId, int iterableSize) {
|
Chain chain,
|
||||||
String ref = loopVar == null ? null : loopVar.getRef();
|
Parameter sourceParameter,
|
||||||
String name = loopVar == null ? null : loopVar.getName();
|
String resultId,
|
||||||
|
int iterableSize) {
|
||||||
|
String ref = sourceParameter == null
|
||||||
|
? null
|
||||||
|
: sourceParameter.getRef();
|
||||||
|
String name = sourceParameter == null
|
||||||
|
? null
|
||||||
|
: sourceParameter.getName();
|
||||||
AtomicBoolean replaced = new AtomicBoolean();
|
AtomicBoolean replaced = new AtomicBoolean();
|
||||||
chain.updateStateSafely(state -> {
|
chain.updateStateSafely(state -> {
|
||||||
ConcurrentHashMap<String, Object> memory =
|
ConcurrentHashMap<String, Object> memory =
|
||||||
@@ -389,18 +535,20 @@ public class LoopNode extends BaseNode {
|
|||||||
*/
|
*/
|
||||||
private LoopInputReference resolveStoredInputReference(
|
private LoopInputReference resolveStoredInputReference(
|
||||||
Chain chain) {
|
Chain chain) {
|
||||||
if (loopVar == null) {
|
Parameter sourceParameter =
|
||||||
|
loopItems != null ? loopItems : loopVar;
|
||||||
|
if (sourceParameter == null) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
Map<String, Object> memory =
|
Map<String, Object> memory =
|
||||||
chain.getExecutionState().getMemory();
|
chain.getExecutionState().getMemory();
|
||||||
String ref = loopVar.getRef();
|
String ref = sourceParameter.getRef();
|
||||||
if (StringUtil.hasText(ref)
|
if (StringUtil.hasText(ref)
|
||||||
&& memory.get(ref)
|
&& memory.get(ref)
|
||||||
instanceof LoopInputReference) {
|
instanceof LoopInputReference) {
|
||||||
return (LoopInputReference) memory.get(ref);
|
return (LoopInputReference) memory.get(ref);
|
||||||
}
|
}
|
||||||
String name = loopVar.getName();
|
String name = sourceParameter.getName();
|
||||||
return StringUtil.hasText(name)
|
return StringUtil.hasText(name)
|
||||||
&& memory.get(name)
|
&& memory.get(name)
|
||||||
instanceof LoopInputReference
|
instanceof LoopInputReference
|
||||||
@@ -418,6 +566,119 @@ public class LoopNode extends BaseNode {
|
|||||||
.set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
|
.set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 解析新结构中的循环次数,并在当前循环上下文中固定结果。
|
||||||
|
*
|
||||||
|
* @param chain 当前工作流
|
||||||
|
* @param context 当前循环上下文
|
||||||
|
* @return 已配置次数;未配置时返回 {@code null}
|
||||||
|
*/
|
||||||
|
private Integer resolveRequestedLoopCount(
|
||||||
|
Chain chain, LoopContext context) {
|
||||||
|
if (loopCount == null) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
if (context.requestedCountResolved) {
|
||||||
|
return context.requestedCount;
|
||||||
|
}
|
||||||
|
int count = parseNumericLoopCount(
|
||||||
|
resolveParameterValue(
|
||||||
|
chain, loopCount, "loop count"));
|
||||||
|
context.requestedCount = count;
|
||||||
|
context.requestedCountResolved = true;
|
||||||
|
return count;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 解析单个循环参数。
|
||||||
|
*
|
||||||
|
* @param chain 当前工作流
|
||||||
|
* @param parameter 待解析参数
|
||||||
|
* @param parameterLabel 错误提示中的参数名称
|
||||||
|
* @return 参数运行值
|
||||||
|
*/
|
||||||
|
private Object resolveParameterValue(
|
||||||
|
Chain chain,
|
||||||
|
Parameter parameter,
|
||||||
|
String parameterLabel) {
|
||||||
|
if (parameter == null
|
||||||
|
|| !StringUtil.hasText(parameter.getName())) {
|
||||||
|
throw new IllegalArgumentException(
|
||||||
|
parameterLabel + " parameter is missing");
|
||||||
|
}
|
||||||
|
Map<String, Object> values =
|
||||||
|
chain.getExecutionState().resolveParameters(
|
||||||
|
this,
|
||||||
|
Collections.singletonList(parameter));
|
||||||
|
return values.get(parameter.getName());
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 固定数组循环的实际迭代次数。
|
||||||
|
*
|
||||||
|
* @param context 当前循环上下文
|
||||||
|
* @param iterableSize 数组实际元素数
|
||||||
|
* @param requestedCount 可选的最大处理数量
|
||||||
|
* @return 本轮实际迭代次数
|
||||||
|
*/
|
||||||
|
private int planIterableIterations(
|
||||||
|
LoopContext context,
|
||||||
|
int iterableSize,
|
||||||
|
Integer requestedCount) {
|
||||||
|
if (context.plannedIterationsResolved) {
|
||||||
|
return context.plannedIterations;
|
||||||
|
}
|
||||||
|
int iterations = requestedCount == null
|
||||||
|
? iterableSize
|
||||||
|
: Math.min(requestedCount, iterableSize);
|
||||||
|
context.plannedIterations = iterations;
|
||||||
|
context.plannedIterationsResolved = true;
|
||||||
|
return iterations;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 固定纯次数循环的实际迭代次数。
|
||||||
|
*
|
||||||
|
* @param context 当前循环上下文
|
||||||
|
* @param iterations 迭代次数
|
||||||
|
*/
|
||||||
|
private void planNumericIterations(
|
||||||
|
LoopContext context, int iterations) {
|
||||||
|
if (!context.plannedIterationsResolved) {
|
||||||
|
context.plannedIterations = iterations;
|
||||||
|
context.plannedIterationsResolved = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 将可迭代输入限制为最多读取指定数量元素。
|
||||||
|
*
|
||||||
|
* @param source 原始可迭代输入
|
||||||
|
* @param limit 最大元素数
|
||||||
|
* @return 单次消费的有界可迭代输入
|
||||||
|
*/
|
||||||
|
private Iterable<?> limitIterable(
|
||||||
|
Iterable<?> source, int limit) {
|
||||||
|
return () -> new Iterator<Object>() {
|
||||||
|
private final Iterator<?> delegate = source.iterator();
|
||||||
|
private int remaining = limit;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean hasNext() {
|
||||||
|
return remaining > 0 && delegate.hasNext();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Object next() {
|
||||||
|
if (!hasNext()) {
|
||||||
|
throw new NoSuchElementException();
|
||||||
|
}
|
||||||
|
remaining--;
|
||||||
|
return delegate.next();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 将集合或数组统一转换为单次消费的 Iterable。
|
* 将集合或数组统一转换为单次消费的 Iterable。
|
||||||
*
|
*
|
||||||
@@ -577,6 +838,22 @@ public class LoopNode extends BaseNode {
|
|||||||
+ "\"");
|
+ "\"");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 创建输入数组类型错误。
|
||||||
|
*
|
||||||
|
* @param loopValue 非法数组输入
|
||||||
|
* @return 参数异常
|
||||||
|
*/
|
||||||
|
private IllegalArgumentException invalidLoopItemsValue(
|
||||||
|
Object loopValue) {
|
||||||
|
String actualType = loopValue == null
|
||||||
|
? "null"
|
||||||
|
: loopValue.getClass().getName();
|
||||||
|
return new IllegalArgumentException(
|
||||||
|
"Loop items must resolve to Iterable or array, but actual type is "
|
||||||
|
+ actualType);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 获取或创建当前节点的 LoopContext 堆栈(每个 LoopNode 实例独立)
|
* 获取或创建当前节点的 LoopContext 堆栈(每个 LoopNode 实例独立)
|
||||||
@@ -856,6 +1133,14 @@ public class LoopNode extends BaseNode {
|
|||||||
long materializationClaimGeneration;
|
long materializationClaimGeneration;
|
||||||
int iterableSize;
|
int iterableSize;
|
||||||
long accumulatedBytes;
|
long accumulatedBytes;
|
||||||
|
/** 是否已固定显式次数输入。 */
|
||||||
|
boolean requestedCountResolved;
|
||||||
|
/** 当前循环固定后的次数输入。 */
|
||||||
|
int requestedCount;
|
||||||
|
/** 是否已固定实际迭代次数。 */
|
||||||
|
boolean plannedIterationsResolved;
|
||||||
|
/** 当前循环固定后的实际迭代次数。 */
|
||||||
|
int plannedIterations;
|
||||||
int expectedReturnCount = 1;
|
int expectedReturnCount = 1;
|
||||||
Set<String> completedBranchIds = new LinkedHashSet<>();
|
Set<String> completedBranchIds = new LinkedHashSet<>();
|
||||||
|
|
||||||
@@ -940,6 +1225,17 @@ public class LoopNode extends BaseNode {
|
|||||||
this.iterableSize = iterableSize;
|
this.iterableSize = iterableSize;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 获取当前循环固定后的实际迭代次数。
|
||||||
|
*
|
||||||
|
* @return 实际迭代次数;尚未规划时返回 {@code null}
|
||||||
|
*/
|
||||||
|
public Integer getPlannedIterations() {
|
||||||
|
return plannedIterationsResolved
|
||||||
|
? plannedIterations
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
public int getExpectedReturnCount() {
|
public int getExpectedReturnCount() {
|
||||||
return expectedReturnCount;
|
return expectedReturnCount;
|
||||||
}
|
}
|
||||||
@@ -968,6 +1264,8 @@ public class LoopNode extends BaseNode {
|
|||||||
|
|
||||||
private final String resultId;
|
private final String resultId;
|
||||||
private final Iterable<?> items;
|
private final Iterable<?> items;
|
||||||
|
private final Parameter sourceParameter;
|
||||||
|
private final boolean externalizeInput;
|
||||||
private final String claimId;
|
private final String claimId;
|
||||||
private final long claimGeneration;
|
private final long claimGeneration;
|
||||||
|
|
||||||
@@ -976,16 +1274,22 @@ public class LoopNode extends BaseNode {
|
|||||||
*
|
*
|
||||||
* @param resultId 唯一结果 ID
|
* @param resultId 唯一结果 ID
|
||||||
* @param items 输入元素
|
* @param items 输入元素
|
||||||
|
* @param sourceParameter 输入来源参数
|
||||||
|
* @param externalizeInput 是否允许替换上游完整输入
|
||||||
* @param claimId 触发器 ID
|
* @param claimId 触发器 ID
|
||||||
* @param claimGeneration 触发器代际
|
* @param claimGeneration 触发器代际
|
||||||
*/
|
*/
|
||||||
private MaterializationPlan(
|
private MaterializationPlan(
|
||||||
String resultId,
|
String resultId,
|
||||||
Iterable<?> items,
|
Iterable<?> items,
|
||||||
|
Parameter sourceParameter,
|
||||||
|
boolean externalizeInput,
|
||||||
String claimId,
|
String claimId,
|
||||||
long claimGeneration) {
|
long claimGeneration) {
|
||||||
this.resultId = resultId;
|
this.resultId = resultId;
|
||||||
this.items = items;
|
this.items = items;
|
||||||
|
this.sourceParameter = sourceParameter;
|
||||||
|
this.externalizeInput = externalizeInput;
|
||||||
this.claimId = claimId;
|
this.claimId = claimId;
|
||||||
this.claimGeneration = claimGeneration;
|
this.claimGeneration = claimGeneration;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,6 +15,7 @@
|
|||||||
*/
|
*/
|
||||||
package com.easyagents.flow.core.parser.impl;
|
package com.easyagents.flow.core.parser.impl;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson.JSONArray;
|
||||||
import com.alibaba.fastjson.JSONObject;
|
import com.alibaba.fastjson.JSONObject;
|
||||||
import com.easyagents.flow.core.chain.Parameter;
|
import com.easyagents.flow.core.chain.Parameter;
|
||||||
import com.easyagents.flow.core.node.LoopNode;
|
import com.easyagents.flow.core.node.LoopNode;
|
||||||
@@ -22,26 +23,49 @@ import com.easyagents.flow.core.parser.BaseNodeParser;
|
|||||||
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 解析显式循环节点的双输入与旧版单输入配置。
|
||||||
|
*/
|
||||||
public class LoopNodeParser extends BaseNodeParser<LoopNode> {
|
public class LoopNodeParser extends BaseNodeParser<LoopNode> {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@inheritDoc}
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public LoopNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) {
|
public LoopNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) {
|
||||||
LoopNode loopNode = new LoopNode();
|
LoopNode loopNode = new LoopNode();
|
||||||
|
|
||||||
// 这里需要设置 id,先设置 id 后, loopNode.setLoopChain(chain); 才能取获取当前节点的 id
|
JSONObject loopInputs = data.getJSONObject("loopInputs");
|
||||||
// loopNode.setId(root.getString("id"));
|
if (loopInputs != null) {
|
||||||
|
loopNode.setLoopCount(
|
||||||
|
parseParameter(loopInputs.getJSONObject("count")));
|
||||||
|
loopNode.setLoopItems(
|
||||||
|
parseParameter(loopInputs.getJSONObject("items")));
|
||||||
|
return loopNode;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 新结构不存在时继续兼容旧版 loopVars[0]。
|
||||||
List<Parameter> loopVars = getParameters(data, "loopVars");
|
List<Parameter> loopVars = getParameters(data, "loopVars");
|
||||||
if (!loopVars.isEmpty()) {
|
if (!loopVars.isEmpty()) {
|
||||||
loopNode.setLoopVar(loopVars.get(0));
|
loopNode.setLoopVar(loopVars.get(0));
|
||||||
}
|
}
|
||||||
|
|
||||||
// JSONArray nodes = chainJSONObject.getJSONArray("nodes");
|
|
||||||
// JSONArray edges = chainJSONObject.getJSONArray("edges");
|
|
||||||
|
|
||||||
// ChainDefinition chain = getChainParser().parse(chainJSONObject, nodes, edges, root);
|
|
||||||
// loopNode.setLoopChain(chain);
|
|
||||||
|
|
||||||
return loopNode;
|
return loopNode;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 将单个参数对象复用基础参数解析逻辑转换为运行时参数。
|
||||||
|
*
|
||||||
|
* @param parameterObject 参数 JSON
|
||||||
|
* @return 参数对象;输入为空时返回 {@code null}
|
||||||
|
*/
|
||||||
|
private Parameter parseParameter(JSONObject parameterObject) {
|
||||||
|
if (parameterObject == null || parameterObject.isEmpty()) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
JSONArray parameters = new JSONArray();
|
||||||
|
parameters.add(parameterObject);
|
||||||
|
List<Parameter> parsed = getParameters(parameters);
|
||||||
|
return parsed.isEmpty() ? null : parsed.get(0);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,95 @@
|
|||||||
|
package com.easyagents.flow.core.test;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson.JSONObject;
|
||||||
|
import com.easyagents.flow.core.chain.DataType;
|
||||||
|
import com.easyagents.flow.core.node.LoopNode;
|
||||||
|
import com.easyagents.flow.core.parser.impl.LoopNodeParser;
|
||||||
|
import org.junit.Assert;
|
||||||
|
import org.junit.Test;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证循环节点新旧输入结构的解析兼容。
|
||||||
|
*/
|
||||||
|
public class LoopNodeParserTest {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证新结构可以同时解析循环次数和输入数组。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldParseExplicitLoopInputs() {
|
||||||
|
JSONObject count = parameter(
|
||||||
|
"count", "Number", "fixed", null, "3");
|
||||||
|
JSONObject items = parameter(
|
||||||
|
"items", "Array<Object>", "ref",
|
||||||
|
"knowledge.documents", null);
|
||||||
|
JSONObject loopInputs = new JSONObject();
|
||||||
|
loopInputs.put("count", count);
|
||||||
|
loopInputs.put("items", items);
|
||||||
|
JSONObject data = new JSONObject();
|
||||||
|
data.put("loopInputs", loopInputs);
|
||||||
|
|
||||||
|
LoopNode loopNode = new LoopNodeParser().doParse(
|
||||||
|
new JSONObject(), data, new JSONObject());
|
||||||
|
|
||||||
|
Assert.assertNotNull(loopNode.getLoopCount());
|
||||||
|
Assert.assertEquals(
|
||||||
|
DataType.Number,
|
||||||
|
loopNode.getLoopCount().getDataType());
|
||||||
|
Assert.assertEquals(
|
||||||
|
"3", loopNode.getLoopCount().getValue());
|
||||||
|
Assert.assertNotNull(loopNode.getLoopItems());
|
||||||
|
Assert.assertEquals(
|
||||||
|
DataType.Array_Object,
|
||||||
|
loopNode.getLoopItems().getDataType());
|
||||||
|
Assert.assertEquals(
|
||||||
|
"knowledge.documents",
|
||||||
|
loopNode.getLoopItems().getRef());
|
||||||
|
Assert.assertNull(loopNode.getLoopVar());
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证旧版 loopVars 第一项继续映射为兼容参数。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldParseLegacyLoopVar() {
|
||||||
|
com.alibaba.fastjson.JSONArray loopVars =
|
||||||
|
new com.alibaba.fastjson.JSONArray();
|
||||||
|
loopVars.add(parameter(
|
||||||
|
"loopVar", "Number", "fixed", null, "2"));
|
||||||
|
JSONObject data = new JSONObject();
|
||||||
|
data.put("loopVars", loopVars);
|
||||||
|
|
||||||
|
LoopNode loopNode = new LoopNodeParser().doParse(
|
||||||
|
new JSONObject(), data, new JSONObject());
|
||||||
|
|
||||||
|
Assert.assertNotNull(loopNode.getLoopVar());
|
||||||
|
Assert.assertEquals("2", loopNode.getLoopVar().getValue());
|
||||||
|
Assert.assertNull(loopNode.getLoopCount());
|
||||||
|
Assert.assertNull(loopNode.getLoopItems());
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 构造参数 JSON。
|
||||||
|
*
|
||||||
|
* @param name 参数名
|
||||||
|
* @param dataType 数据类型
|
||||||
|
* @param refType 引用类型
|
||||||
|
* @param ref 引用路径
|
||||||
|
* @param value 固定值
|
||||||
|
* @return 参数 JSON
|
||||||
|
*/
|
||||||
|
private static JSONObject parameter(
|
||||||
|
String name,
|
||||||
|
String dataType,
|
||||||
|
String refType,
|
||||||
|
String ref,
|
||||||
|
String value) {
|
||||||
|
JSONObject parameter = new JSONObject();
|
||||||
|
parameter.put("name", name);
|
||||||
|
parameter.put("dataType", dataType);
|
||||||
|
parameter.put("refType", refType);
|
||||||
|
parameter.put("ref", ref);
|
||||||
|
parameter.put("value", value);
|
||||||
|
return parameter;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -237,6 +237,110 @@ public class LoopNodeProgressContextTest {
|
|||||||
Arrays.asList("1", "2", "3"), primitiveArrayResult.get("result"));
|
Arrays.asList("1", "2", "3"), primitiveArrayResult.get("result"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证次数和数组同时存在时只遍历数组前 N 项。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldLimitArrayLoopByConfiguredCount() {
|
||||||
|
ChainExecutor executor = createExecutor(
|
||||||
|
createDualInputDefinition(true, true));
|
||||||
|
Map<String, Object> variables = new HashMap<>();
|
||||||
|
variables.put("count", 2);
|
||||||
|
variables.put("items", Arrays.asList("a", "b", "c"));
|
||||||
|
|
||||||
|
Map<String, Object> result = executor.execute(
|
||||||
|
"loop-dual-input-test", variables);
|
||||||
|
|
||||||
|
Assert.assertEquals(
|
||||||
|
Arrays.asList("a", "b"),
|
||||||
|
result.get("result"));
|
||||||
|
Assert.assertEquals(
|
||||||
|
Arrays.asList("a", "b", "c"),
|
||||||
|
result.get("original"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证次数大于数组长度时只处理现有元素。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldStopAtArrayLengthWhenCountIsLarger() {
|
||||||
|
ChainExecutor executor = createExecutor(
|
||||||
|
createDualInputDefinition(true, true));
|
||||||
|
Map<String, Object> variables = new HashMap<>();
|
||||||
|
variables.put("count", 10);
|
||||||
|
variables.put("items", Arrays.asList("a", "b", "c"));
|
||||||
|
|
||||||
|
Map<String, Object> result = executor.execute(
|
||||||
|
"loop-dual-input-short-array-test", variables);
|
||||||
|
|
||||||
|
Assert.assertEquals(
|
||||||
|
Arrays.asList("a", "b", "c"),
|
||||||
|
result.get("result"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证仅配置次数时继续把当前序号作为 loopItem。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldSupportCountOnlyInput() {
|
||||||
|
ChainExecutor executor = createExecutor(
|
||||||
|
createDualInputDefinition(true, false));
|
||||||
|
|
||||||
|
Map<String, Object> result = executor.execute(
|
||||||
|
"loop-count-only-test",
|
||||||
|
Collections.singletonMap("count", 3));
|
||||||
|
|
||||||
|
Assert.assertEquals(
|
||||||
|
Arrays.asList(0, 1, 2),
|
||||||
|
result.get("result"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证配置数组后即使次数合法也拒绝非数组运行值。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldRejectNonArrayItemsWhenCountIsPresent() {
|
||||||
|
ChainExecutor executor = createExecutor(
|
||||||
|
createDualInputDefinition(true, true));
|
||||||
|
Map<String, Object> variables = new HashMap<>();
|
||||||
|
variables.put("count", 2);
|
||||||
|
variables.put("items", "not-an-array");
|
||||||
|
|
||||||
|
assertLoopFailure(
|
||||||
|
executor,
|
||||||
|
"loop-invalid-items-test",
|
||||||
|
variables,
|
||||||
|
RuntimeException.class);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 验证大数组可通过次数限制只处理前 3 项,且上游数组保持完整。
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
public void shouldAllowLargeArrayWhenCountLimitsPrefix() {
|
||||||
|
ChainExecutor executor = createExecutor(
|
||||||
|
createDualInputDefinition(true, true));
|
||||||
|
java.util.List<Integer> items = new java.util.ArrayList<>();
|
||||||
|
for (int index = 0;
|
||||||
|
index < Node.MAX_LOOP_COUNT + 1;
|
||||||
|
index++) {
|
||||||
|
items.add(index);
|
||||||
|
}
|
||||||
|
Map<String, Object> variables = new HashMap<>();
|
||||||
|
variables.put("count", 3);
|
||||||
|
variables.put("items", items);
|
||||||
|
|
||||||
|
Map<String, Object> result = executor.execute(
|
||||||
|
"loop-large-prefix-test", variables);
|
||||||
|
|
||||||
|
Assert.assertEquals(
|
||||||
|
Arrays.asList(0, 1, 2),
|
||||||
|
result.get("result"));
|
||||||
|
Assert.assertEquals(
|
||||||
|
Node.MAX_LOOP_COUNT + 1,
|
||||||
|
((java.util.List<?>) result.get("original")).size());
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 验证显式循环节点接受 300 次,并拒绝 301 次的数值输入。
|
* 验证显式循环节点接受 300 次,并拒绝 301 次的数值输入。
|
||||||
*/
|
*/
|
||||||
@@ -371,6 +475,90 @@ public class LoopNodeProgressContextTest {
|
|||||||
return definition;
|
return definition;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 创建使用新双输入结构的循环定义。
|
||||||
|
*
|
||||||
|
* @param includeCount 是否配置循环次数
|
||||||
|
* @param includeItems 是否配置输入数组
|
||||||
|
* @return 工作流定义
|
||||||
|
*/
|
||||||
|
private ChainDefinition createDualInputDefinition(
|
||||||
|
boolean includeCount, boolean includeItems) {
|
||||||
|
ChainDefinition definition = new ChainDefinition();
|
||||||
|
definition.setId("loop-dual-input-test");
|
||||||
|
|
||||||
|
StartNode startNode = new StartNode();
|
||||||
|
startNode.setId("start");
|
||||||
|
java.util.List<Parameter> inputs =
|
||||||
|
new java.util.ArrayList<>();
|
||||||
|
if (includeCount) {
|
||||||
|
inputs.add(inputParameter("count"));
|
||||||
|
}
|
||||||
|
if (includeItems) {
|
||||||
|
inputs.add(inputParameter("items"));
|
||||||
|
}
|
||||||
|
startNode.setParameters(inputs);
|
||||||
|
|
||||||
|
LoopNode loopNode = new LoopNode();
|
||||||
|
loopNode.setId("loop");
|
||||||
|
loopNode.setName("循环节点");
|
||||||
|
if (includeCount) {
|
||||||
|
Parameter count = new Parameter();
|
||||||
|
count.setName("count");
|
||||||
|
count.setRef("count");
|
||||||
|
count.setRefType(RefType.REF);
|
||||||
|
count.setDataType(DataType.Number);
|
||||||
|
loopNode.setLoopCount(count);
|
||||||
|
}
|
||||||
|
if (includeItems) {
|
||||||
|
Parameter items = new Parameter();
|
||||||
|
items.setName("items");
|
||||||
|
items.setRef("items");
|
||||||
|
items.setRefType(RefType.REF);
|
||||||
|
items.setDataType(DataType.Array);
|
||||||
|
loopNode.setLoopItems(items);
|
||||||
|
}
|
||||||
|
|
||||||
|
ItemCollectorNode collectorNode =
|
||||||
|
new ItemCollectorNode("loop");
|
||||||
|
collectorNode.setId("collector");
|
||||||
|
collectorNode.setParentId("loop");
|
||||||
|
|
||||||
|
Parameter loopOutput = new Parameter();
|
||||||
|
loopOutput.setName("value");
|
||||||
|
loopOutput.setRef("collector.value");
|
||||||
|
loopOutput.setRefType(RefType.REF);
|
||||||
|
loopNode.setOutputDefs(
|
||||||
|
Collections.singletonList(loopOutput));
|
||||||
|
|
||||||
|
EndNode endNode = new EndNode();
|
||||||
|
endNode.setId("end");
|
||||||
|
Parameter result = new Parameter();
|
||||||
|
result.setName("result");
|
||||||
|
result.setRef("loop.value");
|
||||||
|
result.setRefType(RefType.REF);
|
||||||
|
java.util.List<Parameter> outputs =
|
||||||
|
new java.util.ArrayList<>();
|
||||||
|
outputs.add(result);
|
||||||
|
if (includeItems) {
|
||||||
|
Parameter original = new Parameter();
|
||||||
|
original.setName("original");
|
||||||
|
original.setRef("items");
|
||||||
|
original.setRefType(RefType.REF);
|
||||||
|
outputs.add(original);
|
||||||
|
}
|
||||||
|
endNode.setOutputDefs(outputs);
|
||||||
|
|
||||||
|
definition.addNode(startNode);
|
||||||
|
definition.addNode(loopNode);
|
||||||
|
definition.addNode(collectorNode);
|
||||||
|
definition.addNode(endNode);
|
||||||
|
definition.addEdge(edge("e1", "start", "loop"));
|
||||||
|
definition.addEdge(edge("e2", "loop", "collector"));
|
||||||
|
definition.addEdge(edge("e3", "loop", "end"));
|
||||||
|
return definition;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 创建包含两个并行直属循环分支的定义。
|
* 创建包含两个并行直属循环分支的定义。
|
||||||
*
|
*
|
||||||
@@ -707,6 +895,32 @@ public class LoopNodeProgressContextTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 输出当前 loopItem 的测试节点。
|
||||||
|
*/
|
||||||
|
private static class ItemCollectorNode extends BaseNode {
|
||||||
|
private final String loopNodeId;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 创建当前项收集节点。
|
||||||
|
*
|
||||||
|
* @param loopNodeId 父循环节点 ID
|
||||||
|
*/
|
||||||
|
private ItemCollectorNode(String loopNodeId) {
|
||||||
|
this.loopNodeId = loopNodeId;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@inheritDoc}
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public Map<String, Object> execute(Chain chain) {
|
||||||
|
Object item = chain.getState().resolveValue(
|
||||||
|
loopNodeId + ".loopItem");
|
||||||
|
return Collections.singletonMap("value", item);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 输出当前循环序号的测试分支。
|
* 输出当前循环序号的测试分支。
|
||||||
*/
|
*/
|
||||||
|
|||||||
Reference in New Issue
Block a user