SmartBench/CMVR-IOT
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(flow): 重构流程执行拦截器并添加节点名称支持

Browse Source 
- 移除 AbstractFlowMsgPreInterceptor 中的 beforeIntercept 和 afterIntercept 方法实现
- 调整 FlowExceptionInterceptor、FlowLoggingInterceptor、FlowOutputStoreInterceptor 的执行顺序
- 在 FlowExecutionEvent 中新增 nodeName 字段用于存储节点名称
- 修改 FlowItemExecutor 以传递节点名称到执行消息中
- 更新 FlowMsgPreInterceptor 接口移除 beforeIntercept 和 afterIntercept 方法定义
- 新增 FlowAfterInterceptor 处理节点执行后的事件发布
- 新增 FlowBeforeInterceptor 处理节点执行前的事件发布
- 在巡检任务表中添加 map_id 字段并更新相关实体类和映射文件
- 实现巡检流程执行监听器中的节点执行状态推送功能
This commit is contained in:
lixiaolong 2026-06-04 17:37:56 +08:00
parent 4229b07f63
commit c1fe2c582d
14 changed files with 150 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java
View File

@ -47,4 +47,8 @@ public class InspectionTask extends BaseEntity
@ApiModelProperty("任务id")
private String taskConfigId;
@Excel(name = "地图id")
@ApiModelProperty("地图id")
private String mapId;
}

8
cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java
View File

@ -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;
}

26
cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java
View File

@ -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<String, String> 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<InspectionTaskInstance> wrapper = inspectionTaskInstanceService.lambdaQuery()

14
cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml
View File

@ -15,6 +15,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="taskName" column="task_name" />
<result property="taskType" column="task_type" />
<result property="taskConfigId" column="task_config_id" />
<result property="mapId" column="map_id" />
</resultMap>
<resultMap type="com.cmvr.inspection.domain.vo.InspectionTaskVo" id="InspectionTaskVoResult">
@ -30,10 +31,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="createByName" column="create_by_name" />
<result property="updateByName" column="update_by_name" />
<result property="taskConfigId" column="task_config_id" />
<result property="mapId" column="map_id" />
<result property="mapName" column="map_name" />
</resultMap>
<sql id="selectInspectionTaskVo">
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
</sql>
<select id="selectInspectionTaskList" parameterType="InspectionTask" resultMap="InspectionTaskResult">
@ -63,6 +66,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="taskName != null">task_name,</if>
<if test="taskType != null">task_type,</if>
<if test="taskConfigId != null">task_config_id,</if>
<if test="mapId != null">map_id,</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="id != null">#{id},</if>
@ -75,6 +79,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="taskName != null">#{taskName},</if>
<if test="taskType != null">#{taskType},</if>
<if test="taskConfigId != null">#{taskConfigId},</if>
<if test="mapId != null">#{mapId},</if>
</trim>
</insert>
@ -90,6 +95,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="taskName != null">task_name = #{taskName},</if>
<if test="taskType != null">task_type = #{taskType},</if>
<if test="taskConfigId != null">task_config_id = #{taskConfigId},</if>
<if test="mapId != null">map_id = #{mapId},</if>
</trim>
where id = #{id}
</update>
@ -107,15 +113,19 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<select id="selectInspectionTaskVoList" parameterType="InspectionTask" resultMap="InspectionTaskVoResult">
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
<where>
<if test="taskCode != null and taskCode != ''"> and t.task_code = #{taskCode}</if>
<if test="taskName != null and taskName != ''"> and t.task_name like concat('%', #{taskName}, '%')</if>
<if test="taskType != null and taskType != ''"> and t.task_type = #{taskType}</if>
<if test="mapId != null and mapId != ''"> and t.map_id = #{mapId}</if>
</where>
</select>
</mapper>

6
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java
View File

@ -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<Integer> 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

5
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java
View File

@ -47,6 +47,11 @@ public class FlowExecutionEvent {
*/
private String nodeType;
/**
* 节点名称
*/
private String nodeName;
/**
* 任务状态
*/

42
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java
View File

@ -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);
}
}

50
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java Normal file
View File

@ -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<TaskNodeExecuteMessage, TaskNodeExecuteResult> 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;
}
}

37
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java Normal file
View File

@ -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<TaskNodeExecuteMessage, TaskNodeExecuteResult> 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);
}
}

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java
View File

@ -16,7 +16,7 @@ import java.util.function.Function;
@Slf4j
@Component
@Order(1)
@Order(3)
public class FlowExceptionInterceptor extends AbstractFlowMsgPreInterceptor {
private final TaskInstHolder taskInstHolder;

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java
View File

@ -15,7 +15,7 @@ import java.util.function.Function;
@Slf4j
@Component
@Order(3)
@Order(4)
public class FlowLoggingInterceptor extends AbstractFlowMsgPreInterceptor {
private final TaskInstHolder taskInstHolder;

9
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowMsgPreInterceptor.java
View File

@ -18,10 +18,7 @@ public interface FlowMsgPreInterceptor {
*/
default TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message,
Function<TaskNodeExecuteMessage, TaskNodeExecuteResult> 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<TaskNodeExecuteMessage, TaskNodeExecuteResult> next);
void beforeIntercept(TaskNodeExecuteMessage message);
void afterIntercept(TaskNodeExecuteMessage message, TaskNodeExecuteResult result);
}

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java
View File

@ -18,7 +18,7 @@ import java.util.function.Function;
*/
@Slf4j
@Component
@Order(4)
@Order(5)
public class FlowOutputStoreInterceptor extends AbstractFlowMsgPreInterceptor {
private final TaskInstHolder taskInstHolder;

1
sql/inspection_module.sql
View File

@ -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='巡检任务表';