feat(REQ-2752) - 添加流程实例运行结束后，同步最终的状态至 ES

2024-10-18 11:17:11 +08:00 · 2024-10-18 11:17:11 +08:00 · 0bd59ecb9a
commit 0bd59ecb9a
parent 3a16f7e0e7
1 changed files with 94 additions and 0 deletions
--- a/workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/process/SyncToEsProcessEventListener.java
+++ b/workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/process/SyncToEsProcessEventListener.java
@ -0,0 +1,94 @@
+package cn.axzo.workflow.server.controller.listener.process;
+
+import cn.axzo.workflow.core.common.context.ProcessOperationContext;
+import cn.axzo.workflow.core.listener.AbstractBpmnEventListener;
+import cn.axzo.workflow.core.listener.BpmnProcessEventListener;
+import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.flowable.common.engine.api.delegate.event.FlowableEngineEntityEvent;
+import org.flowable.engine.HistoryService;
+import org.flowable.engine.delegate.event.FlowableCancelledEvent;
+import org.flowable.engine.history.HistoricProcessInstance;
+import org.flowable.engine.impl.cfg.ProcessEngineConfigurationImpl;
+import org.flowable.engine.impl.util.CommandContextUtil;
+import org.springframework.context.annotation.Scope;
+import org.springframework.core.Ordered;
+import org.springframework.stereotype.Component;
+
+/**
+ * 流程实例结束后，同步最新数据至 ES
+ *
+ * @author wangli
+ * @since 2024-10-18 11:09
+ */
+@Slf4j
+@Component
+@Scope("prototype")
+@AllArgsConstructor
+public class SyncToEsProcessEventListener extends AbstractBpmnEventListener<ProcessOperationContext> implements BpmnProcessEventListener, Ordered {
+    private final AggregateProcessInstanceService aggregateProcessInstanceService;
+
+    @Override
+    public int getOrder() {
+        return Integer.MIN_VALUE + 2;
+    }
+
+    /**
+     * 流程实例被撤回后回调
+     *
+     * @param event
+     */
+    @Override
+    public void onCancelled(FlowableCancelledEvent event) {
+        syncToEs(event.getProcessInstanceId());
+    }
+
+    /**
+     * 流程实例被驳回后回调
+     *
+     * @param event
+     */
+    @Override
+    public void onRejected(FlowableCancelledEvent event) {
+        syncToEs(event.getProcessInstanceId());
+    }
+
+    /**
+     * 流程实例被中止后回调
+     *
+     * @param event
+     */
+    @Override
+    public void onAborted(FlowableCancelledEvent event) {
+        syncToEs(event.getProcessInstanceId());
+    }
+
+    /**
+     * 流程实例运行完成后回调
+     * <p>
+     * 注意: 完成只是说明流程实例已停止运行
+     *
+     * @param event
+     */
+    @Override
+    public void onCompleted(FlowableEngineEntityEvent event) {
+        syncToEs(event.getProcessInstanceId());
+    }
+
+    private void syncToEs(String processInstanceId) {
+        log.info("SyncToEsProcessEventListener processInstanceId:{}", processInstanceId);
+
+        // 删除指定实例的父子文档
+        aggregateProcessInstanceService.deleteDocumentParentAndChild(processInstanceId);
+        log.info("delete document processInstanceId: {}", processInstanceId);
+
+        ProcessEngineConfigurationImpl processEngineConfiguration = CommandContextUtil.getProcessEngineConfiguration();
+        HistoryService historyService = processEngineConfiguration.getHistoryService();
+        HistoricProcessInstance historicProcessInstance = historyService.createHistoricProcessInstanceQuery()
+                .processInstanceId(processInstanceId)
+                .singleResult();
+        aggregateProcessInstanceService.syncProcessInstance(historicProcessInstance, null);
+        log.info("reInsert document processInstanceId: {}", processInstanceId);
+    }
+}