From c1fe2c582dcdd9777148169aec877742c0a389dd Mon Sep 17 00:00:00 2001 From: lixiaolong <702156524@qq.com> Date: Thu, 4 Jun 2026 17:37:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E9=87=8D=E6=9E=84=E6=B5=81?= =?UTF-8?q?=E7=A8=8B=E6=89=A7=E8=A1=8C=E6=8B=A6=E6=88=AA=E5=99=A8=E5=B9=B6?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=8A=82=E7=82=B9=E5=90=8D=E7=A7=B0=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 AbstractFlowMsgPreInterceptor 中的 beforeIntercept 和 afterIntercept 方法实现 - 调整 FlowExceptionInterceptor、FlowLoggingInterceptor、FlowOutputStoreInterceptor 的执行顺序 - 在 FlowExecutionEvent 中新增 nodeName 字段用于存储节点名称 - 修改 FlowItemExecutor 以传递节点名称到执行消息中 - 更新 FlowMsgPreInterceptor 接口移除 beforeIntercept 和 afterIntercept 方法定义 - 新增 FlowAfterInterceptor 处理节点执行后的事件发布 - 新增 FlowBeforeInterceptor 处理节点执行前的事件发布 - 在巡检任务表中添加 map_id 字段并更新相关实体类和映射文件 - 实现巡检流程执行监听器中的节点执行状态推送功能 --- .../inspection/domain/InspectionTask.java | 4 ++ .../domain/vo/InspectionTaskVo.java | 8 +++ .../InspectionFlowExecutionListener.java | 26 +++++++++- .../inspection/InspectionTaskMapper.xml | 14 +++++- .../flow/runtime/engine/FlowItemExecutor.java | 6 ++- .../runtime/event/FlowExecutionEvent.java | 5 ++ .../AbstractFlowMsgPreInterceptor.java | 42 ---------------- .../interceptor/FlowAfterInterceptor.java | 50 +++++++++++++++++++ .../interceptor/FlowBeforeInterceptor.java | 37 ++++++++++++++ .../interceptor/FlowExceptionInterceptor.java | 2 +- .../interceptor/FlowLoggingInterceptor.java | 2 +- .../interceptor/FlowMsgPreInterceptor.java | 9 +--- .../FlowOutputStoreInterceptor.java | 2 +- sql/inspection_module.sql | 1 + 14 files changed, 150 insertions(+), 58 deletions(-) create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java index 407da2d..f616a49 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java @@ -47,4 +47,8 @@ public class InspectionTask extends BaseEntity @ApiModelProperty("任务id") private String taskConfigId; + @Excel(name = "地图id") + @ApiModelProperty("地图id") + private String mapId; + } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java index a79f76b..940248c 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java @@ -58,4 +58,12 @@ public class InspectionTaskVo extends BaseEntity @ApiModelProperty("任务id") private String taskConfigId; + @Excel(name = "地图id") + @ApiModelProperty("地图id") + private String mapId; + + @Excel(name = "地图名称") + @ApiModelProperty("地图名称") + private String mapName; + } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java index 66b31ee..b459ff2 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java @@ -2,22 +2,46 @@ package com.cmvr.inspection.listener; import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper; import com.cmvr.common.utils.DateUtils; +import com.cmvr.framework.websocket.service.MessagePushService; import com.cmvr.inspection.domain.InspectionTaskInstance; import com.cmvr.inspection.enums.TaskStatusEnum; import com.cmvr.inspection.service.IInspectionTaskInstanceService; import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; import com.cmvr.test.flow.runtime.event.FlowExecutionListener; +import lombok.extern.slf4j.Slf4j; import org.jetbrains.annotations.NotNull; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import java.util.HashMap; +import java.util.Map; + @Component +@Slf4j public class InspectionFlowExecutionListener implements FlowExecutionListener { @Autowired private IInspectionTaskInstanceService inspectionTaskInstanceService; - + @Autowired + private MessagePushService messagePushService; @Override public void onEvent(FlowExecutionEvent event) { + // 节点执行事件 + if (event.getNodeId() != null) { + String info = event.getEventType() == FlowExecutionEvent.EventType.NODE_STARTED ? "开始执行" : "执行完成"; + // 查询巡检任务实例，event中的instId是巡检任务中的taskInsId + InspectionTaskInstance inspectionTaskInstance = inspectionTaskInstanceService.lambdaQuery().eq(InspectionTaskInstance::getTaskInsId, event.getInstId()) + .one(); + String insId = inspectionTaskInstance.getId(); + Map data = new HashMap<>(); + data.put("insId", insId); + data.put("logInfo", info); + try { + messagePushService.pushToChannel("Inspection/TaskInstance", data); + } catch (Exception e) { + log.info("如果没人订阅，则吃掉异常"); + } + + } // 通过 instId 找到巡检任务实例，并修改其状态 // 1.构造更新条件 LambdaQueryChainWrapper wrapper = inspectionTaskInstanceService.lambdaQuery() diff --git a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml index a9fafa6..7192c26 100644 --- a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml +++ b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml @@ -15,6 +15,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + @@ -30,10 +31,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + - select id, create_by, create_time, update_by, update_time, remark, task_code, task_name, task_type, task_config_id from inspection_task + select id, create_by, create_time, update_by, update_time, remark, task_code, task_name, task_type, task_config_id, map_id from inspection_task select t.id, t.create_by, t.create_time, t.update_by, t.update_time, t.remark, - t.task_code, t.task_name, t.task_type, t.task_config_id, + t.task_code, t.task_name, t.task_type, t.task_config_id, t.map_id, + m.map_name, u1.user_name as create_by_name, u2.user_name as update_by_name from inspection_task t left join sys_user u1 on t.create_by = u1.user_name left join sys_user u2 on t.update_by = u2.user_name + + left join inspection_map m on t.map_id = m.id and t.task_code = #{taskCode} and t.task_name like concat('%', #{taskName}, '%') and t.task_type = #{taskType} + and t.map_id = #{mapId} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java index 93c4da6..de6a52a 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java @@ -104,6 +104,7 @@ public class FlowItemExecutor { String instId = rootMessage.getInstId(); String itemId = rootMessage.getItemId(); String nodeId = node.getNodeId(); + String nodeName = node.getNodeName(); log.info("当前执行的是 {} 节点，迭代路径={}", node.getNodeName(), iterations); //注册当前线程 @@ -112,7 +113,7 @@ public class FlowItemExecutor { try { // 创建带上下文的执行消息 TaskNodeExecuteMessage message = buildTaskNodeExecuteMsg( - graph, node, rootMessage, nodeId, iterations + graph, node, rootMessage, nodeId, nodeName, iterations ); // 执行责任链 @@ -157,13 +158,14 @@ public class FlowItemExecutor { @NotNull private static TaskNodeExecuteMessage buildTaskNodeExecuteMsg(FlowGraph graph, FlowNodeWrapper node, - TaskNodeExecuteMessage rootMessage, String nodeId, + TaskNodeExecuteMessage rootMessage, String nodeId, String nodeName, List iterations) { TaskNodeExecuteMessage message = new TaskNodeExecuteMessage(); BeanUtil.copyProperties(rootMessage, message); message.setNodeId(nodeId); message.setNodeType(node.getNodeType().name()); message.setAction(node.getAction()); + message.setNodeName(nodeName); // message.setInputParams(inputParams); message.setGraph(graph); // 只设置 iterations diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java index d9391d6..53b2aac 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java @@ -47,6 +47,11 @@ public class FlowExecutionEvent { */ private String nodeType; + /** + * 节点名称 + */ + private String nodeName; + /** * 任务状态 */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java index 3163d8c..9004277 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java @@ -1,9 +1,6 @@ package com.cmvr.test.flow.runtime.interceptor; -import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; import com.cmvr.test.flow.runtime.event.FlowExecutionEventPublisher; -import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; -import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; /** * 流程消息前置处理 @@ -16,43 +13,4 @@ public abstract class AbstractFlowMsgPreInterceptor implements FlowMsgPreInterce public AbstractFlowMsgPreInterceptor(FlowExecutionEventPublisher eventPublisher) { this.eventPublisher = eventPublisher; } - - @Override - public void beforeIntercept(TaskNodeExecuteMessage message) { - FlowExecutionEvent event = FlowExecutionEvent.builder() - .eventType(FlowExecutionEvent.EventType.NODE_STARTED) - .instId(message.getInstId()) - .taskId(message.getTaskId()) - .itemId(message.getItemId()) - .nodeId(message.getNodeId()) - .nodeType(message.getNodeType()) - .build(); - - eventPublisher.publishEvent(event); - } - - @Override - public void afterIntercept(TaskNodeExecuteMessage message, TaskNodeExecuteResult result) { - FlowExecutionEvent.EventType eventType; - String errorMessage = null; - - if (result != null && result.isSuccess()) { - eventType = FlowExecutionEvent.EventType.NODE_COMPLETED; - } else { - eventType = FlowExecutionEvent.EventType.NODE_FAILED; - errorMessage = result != null ? result.getErrorMsg() : "执行结果为空"; - } - - FlowExecutionEvent event = FlowExecutionEvent.builder() - .eventType(eventType) - .instId(message.getInstId()) - .taskId(message.getTaskId()) - .itemId(message.getItemId()) - .nodeId(message.getNodeId()) - .nodeType(message.getNodeType()) - .errorMessage(errorMessage) - .build(); - - eventPublisher.publishEvent(event); - } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java new file mode 100644 index 0000000..c2455d9 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java @@ -0,0 +1,50 @@ +package com.cmvr.test.flow.runtime.interceptor; + +import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; +import com.cmvr.test.flow.runtime.event.FlowExecutionEventPublisher; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import java.util.function.Function; + +@Slf4j +@Component +@Order(6) +public class FlowAfterInterceptor extends AbstractFlowMsgPreInterceptor{ + + public FlowAfterInterceptor(FlowExecutionEventPublisher eventPublisher) { + super(eventPublisher); + } + + @Override + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { + TaskNodeExecuteResult result = next.apply(message); + FlowExecutionEvent.EventType eventType; + String errorMessage = null; + + if (result != null && result.isSuccess()) { + eventType = FlowExecutionEvent.EventType.NODE_COMPLETED; + } else { + eventType = FlowExecutionEvent.EventType.NODE_FAILED; + errorMessage = result != null ? result.getErrorMsg() : "执行结果为空"; + } + + FlowExecutionEvent event = FlowExecutionEvent.builder() + .eventType(eventType) + .nodeName(message.getNodeName()) + .instId(message.getInstId()) + .nodeName(message.getNodeName()) + .taskId(message.getTaskId()) + .itemId(message.getItemId()) + .nodeId(message.getNodeId()) + .nodeType(message.getNodeType()) + .errorMessage(errorMessage) + .build(); + + eventPublisher.publishEvent(event); + return result; + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java new file mode 100644 index 0000000..0babdfe --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java @@ -0,0 +1,37 @@ +package com.cmvr.test.flow.runtime.interceptor; + +import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; +import com.cmvr.test.flow.runtime.event.FlowExecutionEventPublisher; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import java.util.function.Function; + +@Slf4j +@Component +@Order(1) +public class FlowBeforeInterceptor extends AbstractFlowMsgPreInterceptor{ + + public FlowBeforeInterceptor(FlowExecutionEventPublisher eventPublisher) { + super(eventPublisher); + } + + @Override + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { + FlowExecutionEvent event = FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.NODE_STARTED) + .instId(message.getInstId()) + .taskId(message.getTaskId()) + .itemId(message.getItemId()) + .nodeId(message.getNodeId()) + .nodeName(message.getNodeName()) + .nodeType(message.getNodeType()) + .build(); + + eventPublisher.publishEvent(event); + return next.apply(message); + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java index 429c37a..405a469 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java @@ -16,7 +16,7 @@ import java.util.function.Function; @Slf4j @Component -@Order(1) +@Order(3) public class FlowExceptionInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java index 932054d..4fc24b7 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java @@ -15,7 +15,7 @@ import java.util.function.Function; @Slf4j @Component -@Order(3) +@Order(4) public class FlowLoggingInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowMsgPreInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowMsgPreInterceptor.java index a5fbfd1..783eac0 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowMsgPreInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowMsgPreInterceptor.java @@ -18,10 +18,7 @@ public interface FlowMsgPreInterceptor { */ default TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, Function next) { - beforeIntercept(message); - TaskNodeExecuteResult taskNodeExecuteResult = doIntercept(message, next); - afterIntercept(message, taskNodeExecuteResult); - return taskNodeExecuteResult; + return doIntercept(message, next); } /** @@ -32,8 +29,4 @@ public interface FlowMsgPreInterceptor { */ TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next); - - void beforeIntercept(TaskNodeExecuteMessage message); - - void afterIntercept(TaskNodeExecuteMessage message, TaskNodeExecuteResult result); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java index 8ae82ce..3acd8d2 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java @@ -18,7 +18,7 @@ import java.util.function.Function; */ @Slf4j @Component -@Order(4) +@Order(5) public class FlowOutputStoreInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; diff --git a/sql/inspection_module.sql b/sql/inspection_module.sql index abdf936..a8e7e8b 100644 --- a/sql/inspection_module.sql +++ b/sql/inspection_module.sql @@ -56,6 +56,7 @@ CREATE TABLE `inspection_task` ( `task_name` varchar(100) NOT NULL COMMENT '任务名称', `task_type` char(1) DEFAULT '1' COMMENT '任务类型(1巡检任务 2讲解任务)', `task_config_id` varchar(64) DEFAULT NULL COMMENT '任务配置ID', + `map_id` varchar(64) DEFAULT NULL COMMENT '地图ID', PRIMARY KEY (`id`), UNIQUE KEY `uk_task_code` (`task_code`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='巡检任务表';