From 851dd1be01ba1974e7dae4eb46034b46b798b099 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Wed, 29 Jul 2026 18:05:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E5=BE=AA=E7=8E=AF=E5=8F=8C=E8=BE=93=E5=85=A5=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 支持次数、数组及二者组合的循环执行计划 - 保持旧循环定义和上游数组读取兼容 - 补充解析、边界与前缀遍历测试 --- .../easyagents/flow/core/node/LoopNode.java | 404 +++++++++++++++--- .../flow/core/parser/impl/LoopNodeParser.java | 40 +- .../flow/core/test/LoopNodeParserTest.java | 95 ++++ .../test/LoopNodeProgressContextTest.java | 214 ++++++++++ 4 files changed, 695 insertions(+), 58 deletions(-) create mode 100644 easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeParserTest.java diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java index ff7f2bf..7d5895f 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/LoopNode.java @@ -35,6 +35,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +/** + * 支持纯次数、数组遍历和两者组合限制的显式循环节点。 + */ public class LoopNode extends BaseNode { private static final long serialVersionUID = 1L; @@ -43,16 +46,70 @@ public class LoopNode extends BaseNode { Integer.getInteger( "tinyflow.loop.direct-index.max-items", 64)); + /** 旧版单循环参数。 */ private Parameter loopVar; + /** 可选的循环次数参数。 */ + private Parameter loopCount; + /** 可选的数组输入参数。 */ + private Parameter loopItems; + /** + * 获取旧版循环参数。 + * + * @return 旧版循环参数 + */ public Parameter getLoopVar() { return loopVar; } + /** + * 设置旧版循环参数。 + * + * @param loopVar 旧版循环参数 + */ public void setLoopVar(Parameter 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 public Map execute(Chain chain) { MaterializationPlan[] planHolder = new MaterializationPlan[1]; @@ -164,9 +221,17 @@ public class LoopNode extends BaseNode { boolean directlyIndexed = false; boolean numericLoop = false; Object loopValue = null; + boolean usesExplicitInputs = + loopCount != null || loopItems != null; + Integer requestedCount = usesExplicitInputs + ? resolveRequestedLoopCount(chain, loopContext) + : null; if (loopContext.iterableInputStored) { // 已物化输入不再解析原始参数，避免大集合在每轮反序列化和路径解析。 - shouldLoopCount = loopContext.iterableSize; + shouldLoopCount = planIterableIterations( + loopContext, + loopContext.iterableSize, + requestedCount); storedIterable = true; } else { LoopInputReference storedInput = @@ -178,52 +243,117 @@ public class LoopNode extends BaseNode { storedInput.getItemCount(); loopContext.iterableInputStored = true; loopContext.inputExternalized = true; - shouldLoopCount = - storedInput.getItemCount(); + shouldLoopCount = planIterableIterations( + loopContext, + storedInput.getItemCount(), + requestedCount); storedIterable = true; persistLoopStack(chain, loopStack); - } else { - Map loopVars = - chain.getExecutionState().resolveParameters( - this, - Collections.singletonList( - loopVar)); - loopValue = loopVars.get(loopVar.getName()); + } 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 { + shouldLoopCount = requestedCount == null + ? -1 + : requestedCount; + } + + 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 = toIterableInput(loopValue); if (iterableInput != null) { - int knownSize = knownInputSize(loopValue); - if (knownSize >= 0) { - checkExplicitLoopIterations(chain, knownSize, true); - } - if (knownSize >= 0 - && knownSize <= DIRECT_INDEX_MAX_ITEMS) { - shouldLoopCount = knownSize; - directlyIndexed = true; - } else { - /* - * 每次未完成的物化尝试使用新结果 ID。失锁 owner 的旧分块仅会按 TTL - * 回收，接管者不会复用或删除其部分数据。 - */ - 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, - iterableInput, - loopContext.materializationClaimId, - loopContext.materializationClaimGeneration); - return waitingResult(); - } + int knownSize = knownInputSize(loopValue); + if (knownSize >= 0) { + checkExplicitLoopIterations(chain, knownSize, true); + } + if (knownSize >= 0 + && knownSize <= DIRECT_INDEX_MAX_ITEMS) { + shouldLoopCount = knownSize; + directlyIndexed = true; + } else { + /* + * 每次未完成的物化尝试使用新结果 ID。失锁 owner 的旧分块仅会按 TTL + * 回收，接管者不会复用或删除其部分数据。 + */ + 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, + iterableInput, + loopVar, + true, + loopContext.materializationClaimId, + loopContext.materializationClaimGeneration); + return waitingResult(); + } } else if (loopValue instanceof Number || loopValue instanceof String) { shouldLoopCount = parseNumericLoopCount(loopValue); @@ -231,6 +361,9 @@ public class LoopNode extends BaseNode { } else { throw invalidLoopValue(loopValue); } + } else { + throw new IllegalArgumentException( + "Loop node requires loop count or loop items"); } } checkExplicitLoopIterations(chain, shouldLoopCount, !numericLoop); @@ -329,10 +462,16 @@ public class LoopNode extends BaseNode { "loop-materialization:" + plan.resultId); } context.iterableSize = iterableSize; + context.plannedIterations = iterableSize; + context.plannedIterationsResolved = true; context.iterableInputStored = true; context.materializingInput = false; - context.inputExternalized = externalizeMaterializedInput( - chain, plan.resultId, iterableSize); + context.inputExternalized = plan.externalizeInput + && externalizeMaterializedInput( + chain, + plan.sourceParameter, + plan.resultId, + iterableSize); persistLoopStack(chain, loopStack); return executeLocked(chain, true, null); } @@ -349,9 +488,16 @@ public class LoopNode extends BaseNode { * @return 已替换热状态值时为 {@code true} */ private boolean externalizeMaterializedInput( - Chain chain, String resultId, int iterableSize) { - String ref = loopVar == null ? null : loopVar.getRef(); - String name = loopVar == null ? null : loopVar.getName(); + Chain chain, + Parameter sourceParameter, + String resultId, + int iterableSize) { + String ref = sourceParameter == null + ? null + : sourceParameter.getRef(); + String name = sourceParameter == null + ? null + : sourceParameter.getName(); AtomicBoolean replaced = new AtomicBoolean(); chain.updateStateSafely(state -> { ConcurrentHashMap memory = @@ -389,18 +535,20 @@ public class LoopNode extends BaseNode { */ private LoopInputReference resolveStoredInputReference( Chain chain) { - if (loopVar == null) { + Parameter sourceParameter = + loopItems != null ? loopItems : loopVar; + if (sourceParameter == null) { return null; } Map memory = chain.getExecutionState().getMemory(); - String ref = loopVar.getRef(); + String ref = sourceParameter.getRef(); if (StringUtil.hasText(ref) && memory.get(ref) instanceof LoopInputReference) { return (LoopInputReference) memory.get(ref); } - String name = loopVar.getName(); + String name = sourceParameter.getName(); return StringUtil.hasText(name) && memory.get(name) instanceof LoopInputReference @@ -418,6 +566,119 @@ public class LoopNode extends BaseNode { .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 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() { + 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。 * @@ -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 实例独立） @@ -856,6 +1133,14 @@ public class LoopNode extends BaseNode { long materializationClaimGeneration; int iterableSize; long accumulatedBytes; + /** 是否已固定显式次数输入。 */ + boolean requestedCountResolved; + /** 当前循环固定后的次数输入。 */ + int requestedCount; + /** 是否已固定实际迭代次数。 */ + boolean plannedIterationsResolved; + /** 当前循环固定后的实际迭代次数。 */ + int plannedIterations; int expectedReturnCount = 1; Set completedBranchIds = new LinkedHashSet<>(); @@ -940,6 +1225,17 @@ public class LoopNode extends BaseNode { this.iterableSize = iterableSize; } + /** + * 获取当前循环固定后的实际迭代次数。 + * + * @return 实际迭代次数；尚未规划时返回 {@code null} + */ + public Integer getPlannedIterations() { + return plannedIterationsResolved + ? plannedIterations + : null; + } + public int getExpectedReturnCount() { return expectedReturnCount; } @@ -968,6 +1264,8 @@ public class LoopNode extends BaseNode { private final String resultId; private final Iterable items; + private final Parameter sourceParameter; + private final boolean externalizeInput; private final String claimId; private final long claimGeneration; @@ -976,16 +1274,22 @@ public class LoopNode extends BaseNode { * * @param resultId 唯一结果 ID * @param items 输入元素 + * @param sourceParameter 输入来源参数 + * @param externalizeInput 是否允许替换上游完整输入 * @param claimId 触发器 ID * @param claimGeneration 触发器代际 */ private MaterializationPlan( String resultId, Iterable items, + Parameter sourceParameter, + boolean externalizeInput, String claimId, long claimGeneration) { this.resultId = resultId; this.items = items; + this.sourceParameter = sourceParameter; + this.externalizeInput = externalizeInput; this.claimId = claimId; this.claimGeneration = claimGeneration; } diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/LoopNodeParser.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/LoopNodeParser.java index 3a42639..5cef0cb 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/LoopNodeParser.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/LoopNodeParser.java @@ -15,6 +15,7 @@ */ package com.easyagents.flow.core.parser.impl; +import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.easyagents.flow.core.chain.Parameter; import com.easyagents.flow.core.node.LoopNode; @@ -22,26 +23,49 @@ import com.easyagents.flow.core.parser.BaseNodeParser; import java.util.List; +/** + * 解析显式循环节点的双输入与旧版单输入配置。 + */ public class LoopNodeParser extends BaseNodeParser { + /** + * {@inheritDoc} + */ @Override public LoopNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) { LoopNode loopNode = new LoopNode(); - // 这里需要设置 id，先设置 id 后， loopNode.setLoopChain(chain); 才能取获取当前节点的 id -// loopNode.setId(root.getString("id")); + JSONObject loopInputs = data.getJSONObject("loopInputs"); + if (loopInputs != null) { + loopNode.setLoopCount( + parseParameter(loopInputs.getJSONObject("count"))); + loopNode.setLoopItems( + parseParameter(loopInputs.getJSONObject("items"))); + return loopNode; + } + // 新结构不存在时继续兼容旧版 loopVars[0]。 List loopVars = getParameters(data, "loopVars"); if (!loopVars.isEmpty()) { 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; } + + /** + * 将单个参数对象复用基础参数解析逻辑转换为运行时参数。 + * + * @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 parsed = getParameters(parameters); + return parsed.isEmpty() ? null : parsed.get(0); + } } diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeParserTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeParserTest.java new file mode 100644 index 0000000..9979a5d --- /dev/null +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeParserTest.java @@ -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", "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; + } +} diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeProgressContextTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeProgressContextTest.java index 30ea78d..471c83c 100644 --- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeProgressContextTest.java +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/LoopNodeProgressContextTest.java @@ -237,6 +237,110 @@ public class LoopNodeProgressContextTest { Arrays.asList("1", "2", "3"), primitiveArrayResult.get("result")); } + /** + * 验证次数和数组同时存在时只遍历数组前 N 项。 + */ + @Test + public void shouldLimitArrayLoopByConfiguredCount() { + ChainExecutor executor = createExecutor( + createDualInputDefinition(true, true)); + Map variables = new HashMap<>(); + variables.put("count", 2); + variables.put("items", Arrays.asList("a", "b", "c")); + + Map 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 variables = new HashMap<>(); + variables.put("count", 10); + variables.put("items", Arrays.asList("a", "b", "c")); + + Map 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 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 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 items = new java.util.ArrayList<>(); + for (int index = 0; + index < Node.MAX_LOOP_COUNT + 1; + index++) { + items.add(index); + } + Map variables = new HashMap<>(); + variables.put("count", 3); + variables.put("items", items); + + Map 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 次的数值输入。 */ @@ -371,6 +475,90 @@ public class LoopNodeProgressContextTest { 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 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 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 execute(Chain chain) { + Object item = chain.getState().resolveValue( + loopNodeId + ".loopItem"); + return Collections.singletonMap("value", item); + } + } + /** * 输出当前循环序号的测试分支。 */