diff --git a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionTaskInstanceController.java b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionTaskInstanceController.java index 5b005e2..f9a7b91 100644 --- a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionTaskInstanceController.java +++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionTaskInstanceController.java @@ -2,6 +2,8 @@ package com.cmvr.web.controller.inspection; import java.util.List; import javax.servlet.http.HttpServletResponse; + +import com.cmvr.inspection.domain.dto.InspectionTaskInstanceQuery; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.springframework.security.access.prepost.PreAuthorize; @@ -44,10 +46,10 @@ public class InspectionTaskInstanceController extends BaseController @ApiOperation("查询巡检任务执行实例列表") @PreAuthorize("@ss.hasPermi('inspection:taskInstance:list')") @GetMapping("/list") - public TableDataInfo list(InspectionTaskInstance inspectionTaskInstance) + public TableDataInfo list(InspectionTaskInstanceQuery inspectionTaskInstanceQuery) { startPage(); - List list = inspectionTaskInstanceService.selectInspectionTaskInstanceList(inspectionTaskInstance); + List list = inspectionTaskInstanceService.selectInspectionTaskInstanceList(inspectionTaskInstanceQuery); return getDataTable(list); } @@ -58,9 +60,9 @@ public class InspectionTaskInstanceController extends BaseController @PreAuthorize("@ss.hasPermi('inspection:taskInstance:export')") @Log(title = "巡检任务执行实例", businessType = BusinessType.EXPORT) @PostMapping("/export") - public void export(HttpServletResponse response, InspectionTaskInstance inspectionTaskInstance) + public void export(HttpServletResponse response, InspectionTaskInstanceQuery inspectionTaskInstanceQuery) { - List list = inspectionTaskInstanceService.selectInspectionTaskInstanceList(inspectionTaskInstance); + List list = inspectionTaskInstanceService.selectInspectionTaskInstanceList(inspectionTaskInstanceQuery); ExcelUtil util = new ExcelUtil<>(InspectionTaskInstanceVo.class); util.exportExcel(response, list, "巡检任务执行实例数据"); } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java index 6f2df30..987b66b 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java @@ -4,10 +4,7 @@ import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableId; import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.NoArgsConstructor; +import lombok.*; import com.cmvr.common.annotation.Excel; import com.cmvr.common.core.domain.BaseEntity; @@ -23,6 +20,7 @@ import java.util.Date; @AllArgsConstructor @NoArgsConstructor @EqualsAndHashCode(callSuper = true) +@Builder @ApiModel("巡检任务执行实例") public class InspectionTaskInstance extends BaseEntity { diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionWaypoint.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionWaypoint.java index de29332..a066798 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionWaypoint.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionWaypoint.java @@ -4,10 +4,7 @@ import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableId; import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.NoArgsConstructor; +import lombok.*; import com.cmvr.common.annotation.Excel; import com.cmvr.common.core.domain.BaseEntity; @@ -21,6 +18,7 @@ import com.cmvr.common.core.domain.BaseEntity; @AllArgsConstructor @NoArgsConstructor @EqualsAndHashCode(callSuper = true) +@Builder @ApiModel("巡检点位") public class InspectionWaypoint extends BaseEntity { diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionTaskInstanceQuery.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionTaskInstanceQuery.java new file mode 100644 index 0000000..36807f4 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionTaskInstanceQuery.java @@ -0,0 +1,28 @@ +package com.cmvr.inspection.domain.dto; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.util.List; + +/** + * 巡检任务实例查询入参DTO + */ +@Data +@ApiModel("巡检任务实例查询条件") +public class InspectionTaskInstanceQuery { + + @ApiModelProperty("任务ID") + private String taskId; + + @ApiModelProperty("机器人ID") + private String robotId; + + @ApiModelProperty("单个状态（兼容原有精确查询）") + private Integer status; + + @ApiModelProperty("多状态集合，in查询用：0待执行 1执行中 2已完成 3已取消 4执行失败 5已暂停") + private List statusList; + +} \ No newline at end of file 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 new file mode 100644 index 0000000..66b31ee --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java @@ -0,0 +1,58 @@ +package com.cmvr.inspection.listener; + +import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper; +import com.cmvr.common.utils.DateUtils; +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 org.jetbrains.annotations.NotNull; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class InspectionFlowExecutionListener implements FlowExecutionListener { + @Autowired + private IInspectionTaskInstanceService inspectionTaskInstanceService; + + @Override + public void onEvent(FlowExecutionEvent event) { + // 通过 instId 找到巡检任务实例，并修改其状态 + // 1.构造更新条件 + LambdaQueryChainWrapper wrapper = inspectionTaskInstanceService.lambdaQuery() + .eq(InspectionTaskInstance::getTaskInsId, event.getInstId()); + // 2.要更新的字段 + InspectionTaskInstance updateEntity = getInspectionTaskInstance(); + switch (event.getEventType()) { + case TASK_COMPLETED: + // 设置状态为成功 + updateEntity.setStatus(TaskStatusEnum.SUCCESS.getCode()); + // 正确API：update(更新实体, 条件Wrapper) + inspectionTaskInstanceService.update(updateEntity, wrapper.getWrapper()); + + break; + case TASK_FAILED: + // 设置状态为失败 + updateEntity.setStatus(TaskStatusEnum.FAILED.getCode()); + // 正确API：update(更新实体, 条件Wrapper) + inspectionTaskInstanceService.update(updateEntity, wrapper.getWrapper()); + break; + case NODE_COMPLETED: + // 判断当前任务所有节点是否执行完毕 + break; + } + } + + @NotNull + private static InspectionTaskInstance getInspectionTaskInstance() { + InspectionTaskInstance updateEntity = InspectionTaskInstance.builder() + .endTime(DateUtils.getNowDate()) + .build(); + updateEntity.setUpdateTime(DateUtils.getNowDate()); + return updateEntity; + } + + +} + diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskInstanceMapper.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskInstanceMapper.java index 851bb0f..be7a301 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskInstanceMapper.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskInstanceMapper.java @@ -2,6 +2,7 @@ package com.cmvr.inspection.mapper; import java.util.List; import com.cmvr.inspection.domain.InspectionTaskInstance; +import com.cmvr.inspection.domain.dto.InspectionTaskInstanceQuery; import com.cmvr.inspection.domain.vo.InspectionTaskInstanceVo; import com.github.yulichang.base.MPJBaseMapper; @@ -56,8 +57,8 @@ public interface InspectionTaskInstanceMapper extends MPJBaseMapper selectInspectionTaskInstanceVoList(InspectionTaskInstance inspectionTaskInstance); + List selectInspectionTaskInstanceVoList(InspectionTaskInstanceQuery inspectionTaskInstanceQuery); } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionTaskInstanceService.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionTaskInstanceService.java index 431b36d..771e98e 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionTaskInstanceService.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionTaskInstanceService.java @@ -4,6 +4,7 @@ import java.util.List; import com.baomidou.mybatisplus.extension.service.IService; import com.cmvr.inspection.domain.InspectionTaskInstance; +import com.cmvr.inspection.domain.dto.InspectionTaskInstanceQuery; import com.cmvr.inspection.domain.vo.InspectionTaskInstanceVo; /** * 巡检任务执行实例Service接口 @@ -28,7 +29,7 @@ public interface IInspectionTaskInstanceService extends IService selectInspectionTaskInstanceList(InspectionTaskInstance inspectionTaskInstance); + List selectInspectionTaskInstanceList(InspectionTaskInstanceQuery inspectionTaskInstance); /** * 新增巡检任务执行实例 diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java index 832867e..f07e2b7 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java @@ -3,15 +3,21 @@ package com.cmvr.inspection.service.impl; import java.util.List; import java.util.UUID; +import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.cmvr.common.exception.ServiceException; import com.cmvr.common.utils.SecurityUtils; +import com.cmvr.inspection.domain.InspectionTask; +import com.cmvr.inspection.domain.dto.InspectionTaskInstanceQuery; import com.cmvr.inspection.domain.vo.InspectionTaskInstanceVo; import com.cmvr.inspection.enums.TaskStatusEnum; +import com.cmvr.inspection.service.IInspectionTaskService; import com.cmvr.test.flow.control.FlowControlService; import com.cmvr.test.flow.runtime.engine.FlowTaskRuntimeService; import com.cmvr.test.model.vo.TeTaskExecuteNormalVO; +import com.cmvr.test.service.ITeTaskOrchestrationService; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import com.cmvr.common.utils.DateUtils; import com.cmvr.inspection.mapper.InspectionTaskInstanceMapper; @@ -27,12 +33,19 @@ import com.cmvr.inspection.service.IInspectionTaskInstanceService; @Service public class InspectionTaskInstanceServiceImpl extends ServiceImpl implements IInspectionTaskInstanceService { + @Autowired + private IInspectionTaskService inspectionTaskService; @Autowired private InspectionTaskInstanceMapper inspectionTaskInstanceMapper; @Autowired + private ITeTaskOrchestrationService teTaskOrchestrationService; + + @Autowired + @Lazy private FlowTaskRuntimeService flowTaskRuntimeService; @Autowired + @Lazy private FlowControlService flowControlService; /** @@ -50,13 +63,13 @@ public class InspectionTaskInstanceServiceImpl extends ServiceImpl selectInspectionTaskInstanceList(InspectionTaskInstance inspectionTaskInstance) + public List selectInspectionTaskInstanceList(InspectionTaskInstanceQuery inspectionTaskInstanceQuery) { - return inspectionTaskInstanceMapper.selectInspectionTaskInstanceVoList(inspectionTaskInstance); + return inspectionTaskInstanceMapper.selectInspectionTaskInstanceVoList(inspectionTaskInstanceQuery); } /** @@ -72,6 +85,7 @@ public class InspectionTaskInstanceServiceImpl extends ServiceImpl runParams.put(item.getItemId(), new JSONObject())); // 调用真正的执行 - flowTaskRuntimeService.executeTask(TeTaskExecuteNormalVO.builder().taskId(instance.getTaskInsId()).build()); + String insId = flowTaskRuntimeService.executeTask(TeTaskExecuteNormalVO.builder().taskId(inspectionTask.getTaskConfigId()).runParams(runParams).build()); + update.setTaskInsId(insId); return inspectionTaskInstanceMapper.updateInspectionTaskInstance(update); } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java index 5c28472..245d5a7 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java @@ -161,6 +161,7 @@ public class InspectionTaskServiceImpl extends ServiceImpl - select ti.id, ti.create_by, ti.create_time, ti.update_by, ti.update_time, ti.remark, - ti.task_id, t.task_name, t.task_code, - ti.robot_id, r.robot_name, - r.current_map_id as robot_current_map_id, m.map_name, - ti.status, ti.start_time, ti.end_time, ti.task_ins_id, - u1.nick_name as create_by_name, u2.nick_name as update_by_name + ti.task_id, t.task_name, t.task_code, + ti.robot_id, r.robot_name, + r.current_map_id as robot_current_map_id, m.map_name, + ti.status, ti.start_time, ti.end_time, ti.task_ins_id, + u1.nick_name as create_by_name, u2.nick_name as update_by_name from inspection_task_instance ti left join inspection_task t on ti.task_id = t.id left join inspection_robot r on ti.robot_id = r.id left join inspection_map m on r.current_map_id = m.id - left join sys_user u1 on ti.create_by = u1.user_name - left join sys_user u2 on ti.update_by = u2.user_name + left join sys_user u1 on ti.create_by = u1.user_name + left join sys_user u2 on ti.update_by = u2.user_name - and ti.task_id = #{taskId} - and ti.robot_id = #{robotId} - and ti.status = #{status} + + and ti.task_id = #{taskId} + + + and ti.robot_id = #{robotId} + + + + and ti.status in + + #{item} + + + + + and ti.status = #{status} + order by ti.create_time desc diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java index ebb0566..c509be9 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java @@ -4,6 +4,8 @@ import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.enums.NodeTypeEnum; import com.cmvr.test.enums.TaskStatusEnum; import com.cmvr.test.enums.TerminalStatusEnum; +import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; +import com.cmvr.test.flow.runtime.event.FlowExecutionEventPublisher; import com.cmvr.test.service.ITeNodeInstService; import com.cmvr.test.service.ITeTaskInstService; import lombok.RequiredArgsConstructor; @@ -23,6 +25,7 @@ public class TaskInstHolder { private final ITeNodeInstService nodeInstService; private final TaskContextManager taskContextManager; private final ITeTaskInstService taskInstService; + private final FlowExecutionEventPublisher eventPublisher; public void register(TaskContext context) { taskContextManager.register(context); @@ -52,7 +55,16 @@ public class TaskInstHolder { syncStatus(instId, TaskStatusEnum.SUCCESS); // 如果没有下一个检测项要执行 且是end节点 注销任务上下文 if (pendingItemCount == 0 && nodeType.equalsIgnoreCase(NodeTypeEnum.END.getCode())) { + TaskContext ctx = taskContextManager.get(instId); taskContextManager.unregister(instId); + // 发布任务完成事件 + eventPublisher.publishEvent(FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.TASK_COMPLETED) + .instId(instId) + .taskId(ctx != null ? ctx.getTaskId() : null) + .itemId(itemId) + .status(TaskStatusEnum.SUCCESS) + .build()); } } @@ -63,6 +75,15 @@ public class TaskInstHolder { syncStatus(instId, TaskStatusEnum.FAILED); taskContextManager.unregister(instId); + // 发布任务失败事件 + eventPublisher.publishEvent(FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.TASK_FAILED) + .instId(instId) + .taskId(taskId) + .itemId(itemId) + .status(TaskStatusEnum.FAILED) + .errorMessage(message) + .build()); } public void syncStatus(String instId, TaskStatusEnum status) { @@ -83,6 +104,35 @@ public class TaskInstHolder { ctx.updateTimestamp(); taskInstService.updateStatus(instId, status); + // 发布任务状态变更事件（暂停/终止/恢复） + switch (status) { + case PAUSED: + eventPublisher.publishEvent(FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.TASK_PAUSED) + .instId(instId).taskId(ctx.getTaskId()).itemId(ctx.getItemId()) + .status(status).build()); + break; + case STOPPED: + eventPublisher.publishEvent(FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.TASK_STOPPED) + .instId(instId).taskId(ctx.getTaskId()).itemId(ctx.getItemId()) + .status(status).build()); + break; + default: + break; + } + } + + /** + * 同步状态并标记为恢复操作（发布 TASK_RESUMED 事件） + */ + public void syncStatusAsResumed(String instId) { + TaskContext ctx = taskContextManager.get(instId); + if (ctx == null) return; + eventPublisher.publishEvent(FlowExecutionEvent.builder() + .eventType(FlowExecutionEvent.EventType.TASK_RESUMED) + .instId(instId).taskId(ctx.getTaskId()).itemId(ctx.getItemId()) + .status(TaskStatusEnum.RUNNING).build()); } /** diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java index 7ab789e..70f1099 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java @@ -55,9 +55,7 @@ public class FlowControlService { throw new GlobalException("任务已终止，无需重复操作"); } // 异步终止 - executor.execute(() -> { - edgeSystemService.stopAll(ctx.getTerminalId()); - }); + executor.execute(() -> edgeSystemService.stopAll(ctx.getTerminalId())); // 统一记录日志 + 设置上下文状态 + 数据库状态 taskInstHolder.syncStatus(instId, TaskStatusEnum.STOPPED); @@ -82,9 +80,7 @@ public class FlowControlService { throw new GlobalException("任务已处于暂停状态"); } // 异步停止终端 - executor.execute(() -> { - edgeSystemService.stopAll(ctx.getTerminalId()); - }); + executor.execute(() -> edgeSystemService.stopAll(ctx.getTerminalId())); ctx.setPaused(true); ctx.setTerminalStatus(TerminalStatusEnum.RUNNING); ctx.setStatus(TaskStatusEnum.PAUSED); @@ -197,5 +193,7 @@ public class FlowControlService { ctx.setStatus(TaskStatusEnum.RUNNING); ctx.updateTimestamp(); taskInstHolder.syncStatus(instId, TaskStatusEnum.RUNNING); + // 发布任务恢复事件 + taskInstHolder.syncStatusAsResumed(instId); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java index 7604d38..bfdf645 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java @@ -123,11 +123,11 @@ public class FlowTaskRuntimeEntry implements FlowTaskRuntimeService { public String executeTaskInternal(String terminalId, String taskId, RunModeEnum runMode, JSONObject runParams, Object originalVO) { if (StrUtil.isEmpty(terminalId)) { - throw new GlobalException("终端ID不能为空"); +// throw new GlobalException("终端ID不能为空"); } // 检查终端状态 - if (taskInstHolder.isTerminalLocked(terminalId)) { - throw new GlobalException("当前终端正在执行其他任务，请稍后再试"); + if (terminalId != null && taskInstHolder.isTerminalLocked(terminalId)) { +// throw new GlobalException("当前终端正在执行其他任务，请稍后再试"); } // 查询任务详情（检测项信息） List details = taskOrchestrationService.queryTaskDetail(taskId); 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 new file mode 100644 index 0000000..d9391d6 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java @@ -0,0 +1,81 @@ +package com.cmvr.test.flow.runtime.event; + +import com.cmvr.test.enums.TaskStatusEnum; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * 流程执行事件 + *

+ * 携带流程执行过程中的关键信息，供外部模块监听和处理 + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class FlowExecutionEvent { + + /** + * 事件类型 + */ + private EventType eventType; + + /** + * 任务实例 ID + */ + private String instId; + + /** + * 任务 ID + */ + private String taskId; + + /** + * 当前检测项 ID + */ + private String itemId; + + /** + * 节点 ID（节点级事件时有值） + */ + private String nodeId; + + /** + * 节点类型（节点级事件时有值） + */ + private String nodeType; + + /** + * 任务状态 + */ + private TaskStatusEnum status; + + /** + * 错误信息（失败事件时有值） + */ + private String errorMessage; + + /** + * 事件类型枚举 + */ + public enum EventType { + /** 节点执行完成 */ + NODE_COMPLETED, + /** 节点执行失败 */ + NODE_FAILED, + /** 任务全部执行完成 */ + TASK_COMPLETED, + /** 任务执行失败 */ + TASK_FAILED, + /** 任务被暂停 */ + TASK_PAUSED, + /** 任务被终止 */ + TASK_STOPPED, + /** 任务恢复执行 */ + TASK_RESUMED, + /** 节点开始执行 */ + NODE_STARTED, + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEventPublisher.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEventPublisher.java new file mode 100644 index 0000000..937da08 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEventPublisher.java @@ -0,0 +1,47 @@ +package com.cmvr.test.flow.runtime.event; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.util.Collections; +import java.util.List; + +/** + * 流程执行事件发布器 + *

+ * 自动收集所有 Spring 容器中实现了 {@link FlowExecutionListener} 的 Bean， + * 在流程执行的关键节点发布事件通知。 + */ +@Slf4j +@Component +public class FlowExecutionEventPublisher { + + private final List listeners; + + public FlowExecutionEventPublisher(List listeners) { + this.listeners = listeners != null ? listeners : Collections.emptyList(); + if (!this.listeners.isEmpty()) { + log.info("已注册 {} 个流程执行监听器: {}", this.listeners.size(), + this.listeners.stream().map(l -> l.getClass().getSimpleName()).toArray()); + } + } + + /** + * 发布流程执行事件，通知所有监听器 + * + * @param event 事件对象 + */ + public void publishEvent(FlowExecutionEvent event) { + if (listeners.isEmpty()) { + return; + } + for (FlowExecutionListener listener : listeners) { + try { + listener.onEvent(event); + } catch (Exception e) { + log.error("流程执行监听器 [{}] 处理事件 [{}] 异常: {}", + listener.getClass().getSimpleName(), event.getEventType(), e.getMessage(), e); + } + } + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionListener.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionListener.java new file mode 100644 index 0000000..9e570fd --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionListener.java @@ -0,0 +1,38 @@ +package com.cmvr.test.flow.runtime.event; + +/** + * 流程执行事件监听器 + *

+ * 其他模块可以通过实现此接口来监听流程执行过程中的事件， + * 例如：节点执行完成、任务完成、任务失败、任务暂停/终止/恢复等。 + *

+ * 使用方式：在 Spring 容器中注册一个实现了此接口的 Bean 即可自动生效。 + * 

+ * @Component
+ * public class InspectionFlowExecutionListener implements FlowExecutionListener {
+ *     @Override
+ *     public void onEvent(FlowExecutionEvent event) {
+ *         switch (event.getEventType()) {
+ *             case TASK_COMPLETED:
+ *                 // 处理任务完成逻辑
+ *                 break;
+ *             case TASK_FAILED:
+ *                 // 处理任务失败逻辑
+ *                 break;
+ *         }
+ *     }
+ * }
+ *
+ */ +public interface FlowExecutionListener { + + /** + * 接收流程执行事件 + *

+ * 注意：此方法在流程执行线程中同步调用，请勿执行耗时操作。 + * 如需耗时处理，建议在内部异步执行。 + * + * @param event 流程执行事件 + */ + void onEvent(FlowExecutionEvent event); +} 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 new file mode 100644 index 0000000..3163d8c --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/AbstractFlowMsgPreInterceptor.java @@ -0,0 +1,58 @@ +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; + +/** + * 流程消息前置处理 + */ +public abstract class AbstractFlowMsgPreInterceptor implements FlowMsgPreInterceptor { + + protected final FlowExecutionEventPublisher eventPublisher; + + // 构造注入，子类实例化时由spring注入 + 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/FlowExceptionInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowExceptionInterceptor.java index eef083f..429c37a 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 @@ -5,9 +5,9 @@ import cn.hutool.core.util.ObjUtil; import cn.hutool.core.util.StrUtil; import com.cmvr.common.exception.GlobalException; import com.cmvr.test.flow.context.TaskInstHolder; +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.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; @@ -17,13 +17,17 @@ import java.util.function.Function; @Slf4j @Component @Order(1) -@RequiredArgsConstructor -public class FlowExceptionInterceptor implements FlowMsgPreInterceptor { +public class FlowExceptionInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; + public FlowExceptionInterceptor(FlowExecutionEventPublisher eventPublisher, TaskInstHolder taskInstHolder) { + super(eventPublisher); + this.taskInstHolder = taskInstHolder; + } + @Override - public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { try { return next.apply(message); 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 bfdd943..6c793c8 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 @@ -4,9 +4,9 @@ import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; import com.cmvr.test.flow.runtime.engine.support.FlowNodeParamPreparer; +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.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; @@ -20,13 +20,18 @@ import java.util.function.Function; @Slf4j @Component @Order(2) -@RequiredArgsConstructor -public class FlowInputPrepareInterceptor implements FlowMsgPreInterceptor { +public class FlowInputPrepareInterceptor extends AbstractFlowMsgPreInterceptor { private final FlowNodeParamPreparer paramPreparer; + public FlowInputPrepareInterceptor(FlowExecutionEventPublisher eventPublisher, FlowNodeParamPreparer paramPreparer) { + super(eventPublisher); + this.paramPreparer = paramPreparer; + } + + @Override - public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { FlowGraph graph = message.getGraph(); FlowNodeWrapper node = graph.getNode(message.getNodeId()); 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 d10f4eb..932054d 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 @@ -4,9 +4,9 @@ import cn.hutool.core.util.StrUtil; import com.cmvr.test.enums.NodeTypeEnum; import com.cmvr.test.flow.context.TaskContext; import com.cmvr.test.flow.context.TaskInstHolder; +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.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; @@ -16,13 +16,17 @@ import java.util.function.Function; @Slf4j @Component @Order(3) -@RequiredArgsConstructor -public class FlowLoggingInterceptor implements FlowMsgPreInterceptor { +public class FlowLoggingInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; + public FlowLoggingInterceptor(FlowExecutionEventPublisher eventPublisher, TaskInstHolder taskInstHolder) { + super(eventPublisher); + this.taskInstHolder = taskInstHolder; + } + @Override - public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { long start = System.currentTimeMillis(); log.info("[ {} ]节点开始执行, 参数[ {} ], 循环次数[ {} ]", message.getAction(), message.getInputParams(), message.getLoopNum()); 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 a3418b2..a5fbfd1 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 @@ -16,6 +16,24 @@ public interface FlowMsgPreInterceptor { * @param next 下一个处理函数 * @return 返回最终执行结果 */ - TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, - Function next); + default TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + Function next) { + beforeIntercept(message); + TaskNodeExecuteResult taskNodeExecuteResult = doIntercept(message, next); + afterIntercept(message, taskNodeExecuteResult); + return taskNodeExecuteResult; + } + + /** + * 执行拦截逻辑（子类可重写此方法实现自定义拦截） + * @param message 执行消息 + * @param next 下一个处理函数 + * @return 执行结果 + */ + 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 0a7287b..8ae82ce 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 @@ -3,9 +3,9 @@ package com.cmvr.test.flow.runtime.interceptor; import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.flow.context.TaskContext; import com.cmvr.test.flow.context.TaskInstHolder; +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.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; @@ -19,13 +19,16 @@ import java.util.function.Function; @Slf4j @Component @Order(4) -@RequiredArgsConstructor -public class FlowOutputStoreInterceptor implements FlowMsgPreInterceptor { +public class FlowOutputStoreInterceptor extends AbstractFlowMsgPreInterceptor { private final TaskInstHolder taskInstHolder; + public FlowOutputStoreInterceptor(FlowExecutionEventPublisher eventPublisher, TaskInstHolder taskInstHolder) { + super(eventPublisher); + this.taskInstHolder = taskInstHolder; + } @Override - public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + public TaskNodeExecuteResult doIntercept(TaskNodeExecuteMessage message, Function next) { TaskNodeExecuteResult result = next.apply(message);