diff --git a/cmvr-iot-api/cmvr-iot-edge/cmvr-iot-grpc-client/src/main/java/com/cmvr/edge/client/service/impl/EdgeSpeakerServiceImpl.java b/cmvr-iot-api/cmvr-iot-edge/cmvr-iot-grpc-client/src/main/java/com/cmvr/edge/client/service/impl/EdgeSpeakerServiceImpl.java index 7d5e44c..c1264c3 100644 --- a/cmvr-iot-api/cmvr-iot-edge/cmvr-iot-grpc-client/src/main/java/com/cmvr/edge/client/service/impl/EdgeSpeakerServiceImpl.java +++ b/cmvr-iot-api/cmvr-iot-edge/cmvr-iot-grpc-client/src/main/java/com/cmvr/edge/client/service/impl/EdgeSpeakerServiceImpl.java @@ -1,6 +1,5 @@ package com.cmvr.edge.client.service.impl; -import cmvr.api.Common; import cmvr.api.SpeakerCommand; import cmvr.api.SpeakerServiceGrpc; import cn.hutool.http.HttpRequest; @@ -10,7 +9,6 @@ import com.alibaba.fastjson2.JSONObject; import com.cmvr.edge.client.manage.GrpcServiceManager; import com.cmvr.edge.client.service.EdgeSpeakerService; import com.cmvr.edge.client.utils.EdgeCommonUtil; -import com.google.protobuf.Timestamp; import lombok.RequiredArgsConstructor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java index 069de0b..b962b5b 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java @@ -27,8 +27,6 @@ public class FlowNodeParamPreparer { public JSONObject prepare(FlowGraph graph, FlowNodeWrapper node, - String startNodeId, - List startInputDefs, TaskNodeExecuteMessage rootMessage) { @@ -50,21 +48,17 @@ public class FlowNodeParamPreparer { List path = param.getQuote().subList(2, param.getQuote().size()); // 取值路径 JSONObject source; + // 引用的参数是上游的输入还是输出 + // 2.1 引用上游输入参数 if ("input".equalsIgnoreCase(param.getQuoteType())) { - if (refNodeId.equals(startNodeId)) { - // Start 节点:优先取 runParams(实际运行时输入) - source = ctx.getRunParams(); - if (source == null || source.isEmpty()) { - // 如果运行时没有,就回退到定义的 startInputDefs - source = toJson(startInputDefs); - } - } else { - // 普通节点:取定义参数 - source = toJson(graph.getNodeParamsDefs().get(refNodeId)); + source = ctx.getRunParams(); + JSONObject def = toJson(graph.getNodeParamsDefs().get(refNodeId)); + for (String key : def.keySet()) { + source.put(key, def.get(key)); } } else if ("output".equalsIgnoreCase(param.getQuoteType())) { // 2.2 引用上游的 output 参数(运行时实际执行结果) - // 这里支持循环:用 nodeId + iterations 作为 key,避免覆盖 + // 用 nodeId + iterations 作为 key,避免覆盖 source = ctx.getNodeOutput(refNodeId, rootMessage.getIterations()); } else { throw new GlobalException("参数应用类型出错!quoteType:" + param.getQuoteType()); @@ -166,7 +160,6 @@ public class FlowNodeParamPreparer { /** * 根据迭代路径(iterations)逐层解析嵌套集合 - * 例子: * - iterations = [] → 返回最外层对象 * - iterations = [2] → 返回 list[1] * - iterations = [2,3] → 返回 list[1][2] diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java index 49fa8fb..bfdd943 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java @@ -3,7 +3,6 @@ package com.cmvr.test.flow.runtime.interceptor; import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; -import com.cmvr.test.flow.builder.FlowParamDef; import com.cmvr.test.flow.runtime.engine.support.FlowNodeParamPreparer; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; @@ -12,7 +11,6 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; -import java.util.List; import java.util.function.Function; /** @@ -32,17 +30,9 @@ public class FlowInputPrepareInterceptor implements FlowMsgPreInterceptor { Function next) { FlowGraph graph = message.getGraph(); FlowNodeWrapper node = graph.getNode(message.getNodeId()); - String startNodeId = graph.isSub() ? graph.findSubStartNodeId().getNodeId(): graph.findStartNodeId(); - List startInputDefs = graph.getStartInputParamsDefs(); // 调用已有的 prepare 方法 - JSONObject inputParams = paramPreparer.prepare( - graph, - node, - startNodeId, - startInputDefs, - message - ); + JSONObject inputParams = paramPreparer.prepare(graph, node, message); message.setInputParams(inputParams); log.debug("[InputPrepare] 节点 {} 准备输入参数: {}", message.getNodeId(), inputParams);