From 876517f821aa43a39f9f0950604ab947e97d0a76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 31 Aug 2026 15:54:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M28=20=E6=94=AF=E6=8C=81=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E6=B5=81=E5=A4=9A=E5=85=A5=E8=BE=B9=E6=B1=87=E8=81=9A?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/easyagents/flow/core/chain/Chain.java | 11 +- .../com/easyagents/flow/core/chain/Node.java | 30 ++ .../flow/core/chain/NodeJoinMode.java | 60 +++ .../flow/core/parser/BaseNodeParser.java | 5 + .../flow/core/test/NodeJoinModeTest.java | 407 ++++++++++++++++++ 5 files changed, 510 insertions(+), 3 deletions(-) create mode 100644 easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java create mode 100644 easy-agents-flow/src/test/java/com/easyagents/flow/core/test/NodeJoinModeTest.java diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java index b04db36..a539521 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java @@ -634,7 +634,8 @@ public class Chain { NodeStateField .EXECUTION_ATTEMPT_KEY); } - if (node.getCondition() == null) { + if (node.getJoinMode() == NodeJoinMode.ANY + && node.getCondition() == null) { s.recordTrigger(triggerEdgeId); fields.add(NodeStateField.TRIGGER_COUNT); fields.add(NodeStateField.TRIGGER_EDGE_IDS); @@ -795,7 +796,7 @@ public class Chain { private boolean shouldSkipNode(Node node, String edgeId) { NodeCondition condition = node.getCondition(); - if (condition == null) { + if (node.getJoinMode() == NodeJoinMode.ANY && condition == null) { return false; } return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS, () -> { @@ -805,7 +806,11 @@ public class Chain { }); Map prevResult = Collections.emptyMap(); - boolean shouldSkipNode = !condition.check(this, newState, prevResult); + boolean joinPending = node.getJoinMode() == NodeJoinMode.ALL + && !newState.isUpstreamFullyExecuted(); + boolean shouldSkipNode = joinPending + || (condition != null + && !condition.check(this, newState, prevResult)); if (shouldSkipNode) { updateStateSafely(state -> { return state.addUncheckedNodeId(node.id) diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java index 643a1da..a61486a 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java @@ -42,6 +42,7 @@ public abstract class Node implements Serializable { protected NodeCondition condition; protected NodeValidator validator; + protected NodeJoinMode joinMode = NodeJoinMode.ANY; // 循环执行相关属性 protected boolean loopEnable = false; // 是否启用循环执行 @@ -70,6 +71,10 @@ public abstract class Node implements Serializable { } public void setParentId(String parentId) { + if (StringUtil.hasText(parentId) && getJoinMode() == NodeJoinMode.ALL) { + throw new IllegalArgumentException( + "joinMode 'all' is not supported for loop child nodes"); + } this.parentId = parentId; } @@ -121,6 +126,31 @@ public abstract class Node implements Serializable { this.validator = validator; } + /** + * 获取节点的直接入边汇聚模式。 + * + *

旧序列化对象缺少该字段时返回 {@link NodeJoinMode#ANY}。

+ * + * @return 汇聚模式 + */ + public NodeJoinMode getJoinMode() { + return joinMode == null ? NodeJoinMode.ANY : joinMode; + } + + /** + * 设置节点的直接入边汇聚模式。 + * + * @param joinMode 汇聚模式 + */ + public void setJoinMode(NodeJoinMode joinMode) { + NodeJoinMode resolved = joinMode == null ? NodeJoinMode.ANY : joinMode; + if (resolved == NodeJoinMode.ALL && StringUtil.hasText(parentId)) { + throw new IllegalArgumentException( + "joinMode 'all' is not supported for loop child nodes"); + } + this.joinMode = resolved; + } + // protected void addOutwardEdge(Edge edge) { // if (this.outwardEdges == null) { // this.outwardEdges = new ArrayList<>(); diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java new file mode 100644 index 0000000..24a851f --- /dev/null +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java @@ -0,0 +1,60 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.gnu.org/licenses/lgpl-3.0.txt + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.easyagents.flow.core.chain; + +import java.util.Locale; + +/** + * 多入边节点的触发汇聚模式。 + */ +public enum NodeJoinMode { + + /** 任意一条直接入边到达即可执行。 */ + ANY("any"), + /** 全部直接入边到达后才执行。 */ + ALL("all"); + + private final String value; + + NodeJoinMode(String value) { + this.value = value; + } + + public String getValue() { + return value; + } + + /** + * 按工作流 JSON 值解析汇聚模式。 + * + * @param value 配置值 + * @return 汇聚模式 + * @throws IllegalArgumentException 配置为空或不受支持 + */ + public static NodeJoinMode ofValue(String value) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException("joinMode must be 'any' or 'all'"); + } + String normalized = value.trim().toLowerCase(Locale.ROOT); + for (NodeJoinMode mode : values()) { + if (mode.value.equals(normalized)) { + return mode; + } + } + throw new IllegalArgumentException( + "Unsupported joinMode: " + value + "; expected 'any' or 'all'"); + } +} diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java index cbe5eab..93c522e 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java @@ -21,6 +21,7 @@ import com.alibaba.fastjson.JSONObject; import com.easyagents.flow.core.chain.DataType; import com.easyagents.flow.core.chain.JsCodeCondition; import com.easyagents.flow.core.chain.Node; +import com.easyagents.flow.core.chain.NodeJoinMode; import com.easyagents.flow.core.chain.Parameter; import com.easyagents.flow.core.chain.RefType; import com.easyagents.flow.core.node.BaseNode; @@ -123,6 +124,10 @@ public abstract class BaseNodeParser implements NodeParser parseEndNode(parser, "first", null)); + assertInvalidJoinMode(() -> parseEndNode(parser, "", null)); + assertInvalidJoinMode(() -> parseEndNode(parser, "all", "loop")); + + ProbeJoinNode node = new ProbeJoinNode(new AtomicInteger()); + node.setJoinMode(NodeJoinMode.ALL); + assertInvalidJoinMode(() -> node.setParentId("loop")); + } + + @Test + public void shouldWaitForEveryInboundEdgeBeforeExecutingAndCheckingCondition() + throws Exception { + JoinFixture fixture = createJoinFixture(NodeJoinMode.ALL, true); + String instanceId = null; + try { + instanceId = fixture.executor.executeAsync( + fixture.definition.getId(), Collections.emptyMap()); + Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS)); + Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS)); + + String currentInstanceId = instanceId; + await(() -> hasTriggerEdge( + fixture.nodeStateRepository, + currentInstanceId, + "join", + "a-join")); + Assert.assertEquals(0, fixture.joinExecutions.get()); + Assert.assertEquals(0, fixture.conditionChecks.get()); + + fixture.releaseBranchB.countDown(); + ChainState finalState = awaitTerminal( + fixture.chainStateRepository, instanceId); + + Assert.assertEquals(1, fixture.joinExecutions.get()); + Assert.assertEquals(1, fixture.conditionChecks.get()); + Assert.assertEquals("A+B", finalState.getExecuteResult().get("combined")); + Assert.assertEquals(Boolean.TRUE, finalState.getExecuteResult().get("sawBoth")); + } finally { + fixture.releaseBranchB.countDown(); + fixture.scheduler.shutdown(); + } + } + + @Test + public void shouldKeepAnyModeFirstArrivalBehavior() throws Exception { + JoinFixture fixture = createJoinFixture(NodeJoinMode.ANY, false); + try { + String instanceId = fixture.executor.executeAsync( + fixture.definition.getId(), Collections.emptyMap()); + Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS)); + Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS)); + + await(() -> fixture.joinExecutions.get() > 0); + ChainState finalState = awaitTerminal( + fixture.chainStateRepository, instanceId); + + Assert.assertEquals(1, fixture.joinExecutions.get()); + Assert.assertEquals(Boolean.FALSE, finalState.getExecuteResult().get("sawBoth")); + } finally { + fixture.releaseBranchB.countDown(); + fixture.scheduler.shutdown(); + } + } + + @Test + public void shouldKeepDefaultAnyRetryAndLoopBehavior() throws Exception { + ScheduledExecutorService schedulerPool = + Executors.newSingleThreadScheduledExecutor(); + ExecutorService workerPool = Executors.newFixedThreadPool(3); + TriggerScheduler scheduler = new TriggerScheduler( + new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L); + AtomicInteger executions = new AtomicInteger(); + ChainDefinition definition = new ChainDefinition(); + definition.setId("join-mode-retry-loop"); + StartNode start = new StartNode(); + start.setId("start"); + RetryLoopNode worker = new RetryLoopNode(executions); + worker.setId("worker"); + worker.setRetryEnable(true); + worker.setMaxRetryCount(1); + worker.setRetryIntervalMs(0L); + worker.setLoopEnable(true); + worker.setMaxLoopCount(2); + worker.setLoopIntervalMs(0L); + EndNode end = endNode("worker", "count", "count"); + definition.addNode(start); + definition.addNode(worker); + definition.addNode(end); + definition.addEdge(edge("start-worker", "start", "worker")); + definition.addEdge(edge("worker-end", "worker", "end")); + + ChainExecutor executor = new ChainExecutor( + ignored -> definition, + new InMemoryChainStateRepository(), + new InMemoryNodeStateRepository(), + scheduler); + try { + Map result = executor.execute( + definition.getId(), Collections.emptyMap(), 5L, TimeUnit.SECONDS); + + Assert.assertEquals(NodeJoinMode.ANY, worker.getJoinMode()); + Assert.assertEquals(3, executions.get()); + Assert.assertEquals(3, result.get("count")); + } finally { + scheduler.shutdown(); + } + } + + private JoinFixture createJoinFixture( + NodeJoinMode joinMode, boolean withCondition) { + JoinFixture fixture = new JoinFixture(); + fixture.schedulerPool = Executors.newScheduledThreadPool(2); + fixture.workerPool = Executors.newFixedThreadPool(4); + fixture.scheduler = new TriggerScheduler( + new InMemoryTriggerStore(), + fixture.schedulerPool, + fixture.workerPool, + 1000L); + fixture.chainStateRepository = new InMemoryChainStateRepository(); + fixture.nodeStateRepository = new InMemoryNodeStateRepository(); + fixture.definition = new ChainDefinition(); + fixture.definition.setId("join-mode-" + joinMode.getValue()); + + StartNode start = new StartNode(); + start.setId("start"); + BranchNode branchA = new BranchNode( + "A", fixture.branchACompleted, null, null); + branchA.setId("a"); + BranchNode branchB = new BranchNode( + "B", null, fixture.branchBStarted, fixture.releaseBranchB); + branchB.setId("b"); + ProbeJoinNode join = new ProbeJoinNode(fixture.joinExecutions); + join.setId("join"); + join.setJoinMode(joinMode); + if (withCondition) { + join.setCondition(new BothOutputsCondition(fixture.conditionChecks)); + } + EndNode end = endNode("join", "combined", "combined"); + end.addOutputDef(outputRef("sawBoth", "join.sawBoth")); + + fixture.definition.addNode(start); + fixture.definition.addNode(branchA); + fixture.definition.addNode(branchB); + fixture.definition.addNode(join); + fixture.definition.addNode(end); + fixture.definition.addEdge(edge("start-a", "start", "a")); + fixture.definition.addEdge(edge("start-b", "start", "b")); + fixture.definition.addEdge(edge("a-join", "a", "join")); + fixture.definition.addEdge(edge("b-join", "b", "join")); + fixture.definition.addEdge(edge("join-end", "join", "end")); + + fixture.executor = new ChainExecutor( + ignored -> fixture.definition, + fixture.chainStateRepository, + fixture.nodeStateRepository, + fixture.scheduler); + return fixture; + } + + private Node parseEndNode( + ChainParser parser, String joinMode, String parentId) { + JSONObject data = new JSONObject(); + if (joinMode != null) { + data.put("joinMode", joinMode); + } + JSONObject nodeJson = new JSONObject(); + nodeJson.put("id", "end"); + nodeJson.put("type", "endNode"); + nodeJson.put("parentId", parentId); + nodeJson.put("data", data); + return new EndNodeParser().parse( + nodeJson, new JSONObject(), parser); + } + + private void assertInvalidJoinMode(Runnable action) { + try { + action.run(); + Assert.fail("Expected invalid join mode"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("joinMode")); + } + } + + private static boolean hasTriggerEdge( + InMemoryNodeStateRepository repository, + String instanceId, + String nodeId, + String edgeId) { + NodeState state = repository.load(instanceId, nodeId); + return state != null && state.getTriggerEdgeIds().contains(edgeId); + } + + private static ChainState awaitTerminal( + InMemoryChainStateRepository repository, + String instanceId) throws Exception { + await(() -> { + ChainState state = repository.load(instanceId); + return state != null + && state.getStatus() != null + && state.getStatus().isTerminal(); + }); + ChainState state = repository.load(instanceId); + Assert.assertEquals(ChainStatus.SUCCEEDED, state.getStatus()); + return state; + } + + private static void await(BooleanSupplier condition) throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3L); + while (!condition.getAsBoolean() && System.nanoTime() < deadline) { + Thread.sleep(10L); + } + Assert.assertTrue("condition was not met before timeout", condition.getAsBoolean()); + } + + private static EndNode endNode( + String sourceNodeId, String sourceName, String outputName) { + EndNode end = new EndNode(); + end.setId("end"); + end.addOutputDef(outputRef( + outputName, sourceNodeId + "." + sourceName)); + return end; + } + + private static Parameter outputRef(String name, String ref) { + Parameter parameter = new Parameter(); + parameter.setName(name); + parameter.setRef(ref); + parameter.setRefType(RefType.REF); + return parameter; + } + + private static Edge edge(String id, String source, String target) { + Edge edge = new Edge(); + edge.setId(id); + edge.setSource(source); + edge.setTarget(target); + return edge; + } + + private static final class BranchNode extends BaseNode { + private final String value; + private final CountDownLatch completed; + private final CountDownLatch started; + private final CountDownLatch release; + + private BranchNode( + String value, + CountDownLatch completed, + CountDownLatch started, + CountDownLatch release) { + this.value = value; + this.completed = completed; + this.started = started; + this.release = release; + } + + @Override + public Map execute(Chain chain) { + if (started != null) { + started.countDown(); + } + if (release != null) { + try { + if (!release.await(3L, TimeUnit.SECONDS)) { + throw new IllegalStateException("branch release timed out"); + } + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("branch interrupted", error); + } + } + if (completed != null) { + completed.countDown(); + } + return Collections.singletonMap("value", value); + } + } + + private static final class ProbeJoinNode extends BaseNode { + private final AtomicInteger executions; + + private ProbeJoinNode(AtomicInteger executions) { + this.executions = executions; + } + + @Override + public Map execute(Chain chain) { + executions.incrementAndGet(); + Object a = chain.getExecutionState().getMemory().get("a.value"); + Object b = chain.getExecutionState().getMemory().get("b.value"); + Map result = new HashMap<>(); + result.put("combined", String.valueOf(a) + "+" + String.valueOf(b)); + result.put("sawBoth", a != null && b != null); + return result; + } + } + + private static final class BothOutputsCondition implements NodeCondition { + private static final long serialVersionUID = 1L; + private final AtomicInteger checks; + + private BothOutputsCondition(AtomicInteger checks) { + this.checks = checks; + } + + @Override + public boolean check( + Chain chain, + NodeState context, + Map executeResult) { + checks.incrementAndGet(); + Map memory = chain.getExecutionState().getMemory(); + return memory.containsKey("a.value") && memory.containsKey("b.value"); + } + } + + private static final class RetryLoopNode extends BaseNode { + private final AtomicInteger executions; + + private RetryLoopNode(AtomicInteger executions) { + this.executions = executions; + } + + @Override + public Map execute(Chain chain) { + int count = executions.incrementAndGet(); + if (count == 1) { + throw new IllegalStateException("retry once"); + } + return Collections.singletonMap("count", count); + } + } + + private static final class JoinFixture { + private final CountDownLatch branchACompleted = new CountDownLatch(1); + private final CountDownLatch branchBStarted = new CountDownLatch(1); + private final CountDownLatch releaseBranchB = new CountDownLatch(1); + private final AtomicInteger joinExecutions = new AtomicInteger(); + private final AtomicInteger conditionChecks = new AtomicInteger(); + private ScheduledExecutorService schedulerPool; + private ExecutorService workerPool; + private TriggerScheduler scheduler; + private InMemoryChainStateRepository chainStateRepository; + private InMemoryNodeStateRepository nodeStateRepository; + private ChainDefinition definition; + private ChainExecutor executor; + } +}