wangli/workflow-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/REQ-2752' into feature/merge_2924_2752

Browse Source
This commit is contained in:
wangli 2024-10-16 17:10:12 +08:00
commit a32279acc4
9 changed files with 109 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/cmd/CustomTransferUserTaskCmd.java
View File

@ -114,7 +114,7 @@ public class CustomTransferUserTaskCmd extends AbstractCommand<Void> implements
task.setScopeType("TRANSFER");
Authentication.setAuthenticatedUserId(originTaskAssignee.buildAssigneeId());
addComment(commandContext, task, COMMENT_TYPE_OPERATION_DESC, "转交给" + targetTaskAssignee.getAssignerName());
addComment(commandContext, task, COMMENT_TYPE_OPERATION_DESC, "转交给" + targetTaskAssignee.getAssignerName() + additionalOpeDesc);
addComment(commandContext, task, COMMENT_TYPE_ADVICE, advice);
Authentication.setAuthenticatedUserId(null);

6
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/es/model/ProcessTaskDocument.java
View File

@ -24,6 +24,12 @@ public class ProcessTaskDocument {
@IndexId(type = IdType.CUSTOMIZE)
private String taskId;
/**
* 任务归属的实例 ID
*/
@IndexField(fieldType = FieldType.KEYWORD)
private String processInstanceId;
/**
* 任务定义 KEY,对应节点 ID
*/

2
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/es/service/EsProcessTaskService.java
View File

@ -42,4 +42,6 @@ public interface EsProcessTaskService {
Integer insertBatch(String routing, String parentId, Collection<ProcessTaskDocument> processTaskDocuments);
List<ProcessTaskDocument> search(TaskSearchReqDTO dto);
Integer delete(String routing, String processInstanceId);
}

17
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/es/service/aggregation/AggregateProcessInstanceService.java
View File

@ -143,4 +143,21 @@ public class AggregateProcessInstanceService {
List<ProcessInstanceDocumentVO> vos = BeanMapper.copyList(searchResult.getList(), ProcessInstanceDocumentVO.class);
return new BpmPageResult<>(vos, searchResult.getTotal());
}
/**
* 删除指定流程实例的父子文档
*
* @param processInstanceId
* @return
*/
public Boolean deleteDocumentParentAndChild(String processInstanceId) {
try {
esProcessInstanceService.delete(processInstanceId);
aggregateProcessTaskService.deleteByParentId(processInstanceId);
} catch (Exception e) {
log.warn("删除文档失败：{}", e.getMessage(), e);
return false;
}
return true;
}
}

9
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/es/service/aggregation/AggregateProcessTaskService.java
View File

@ -15,9 +15,12 @@ import cn.axzo.workflow.es.service.EsProcessTaskService;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import com.google.common.collect.Lists;
import io.swagger.models.auth.In;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.maven.artifact.versioning.DefaultArtifactVersion;
import org.dromara.easyes.core.conditions.select.LambdaEsQueryWrapper;
import org.dromara.easyes.core.toolkit.FieldUtils;
import org.flowable.bpmn.model.BpmnModel;
import org.flowable.bpmn.model.FlowElement;
import org.flowable.engine.history.HistoricProcessInstance;
@ -82,6 +85,7 @@ public class AggregateProcessTaskService {
logs.forEach(log -> {
ProcessTaskDocument processTaskDocument = new ProcessTaskDocument();
processTaskDocument.setTaskId(log.getTaskId());
processTaskDocument.setProcessInstanceId(hpi.getId());
processTaskDocument.setTaskDefinitionKey(log.getActivityId());
processTaskDocument.setTaskName(log.getActivityName());
processTaskDocument.setTaskStatus(log.getStatus());
@ -159,6 +163,7 @@ public class AggregateProcessTaskService {
ProcessTaskDocument processTaskDocument = new ProcessTaskDocument();
processTaskDocument.setTaskId(task.getId());
processTaskDocument.setProcessInstanceId(hpi.getId());
processTaskDocument.setTaskDefinitionKey(task.getTaskDefinitionKey());
processTaskDocument.setTaskName(task.getName());
processTaskDocument.setTaskStatus(extTask.getStatus());
@ -240,4 +245,8 @@ public class AggregateProcessTaskService {
public List<ProcessTaskDocumentVO> search(TaskSearchReqDTO dto) {
return BeanMapper.copyList(esProcessTaskService.search(dto), ProcessTaskDocumentVO.class);
}
public Integer deleteByParentId(String processInstanceId) {
return esProcessTaskService.delete(ES_FIXED_ROUTING, processInstanceId);
}
}

8
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/es/service/impl/EsProcessTaskServiceImpl.java
View File

@ -72,4 +72,12 @@ public class EsProcessTaskServiceImpl implements EsProcessTaskService {
;
return esProcessTaskMapper.selectList(wrapper);
}
@Override
public Integer delete(String routing, String processInstanceId) {
LambdaEsQueryWrapper<ProcessTaskDocument> wrapper = new LambdaEsQueryWrapper<ProcessTaskDocument>()
.eq(FieldUtils.val(ProcessTaskDocument::getProcessInstanceId), processInstanceId)
.routing(ES_FIXED_ROUTING);
return esProcessTaskMapper.delete(wrapper);
}
}

56
workflow-engine-elasticsearch/src/main/java/cn/axzo/workflow/listener/SyncEsTaskEntityEventHandle.java
View File

@ -1,56 +0,0 @@
package cn.axzo.workflow.listener;
import cn.axzo.workflow.core.engine.listener.entity.EntityEventHandle;
import cn.axzo.workflow.es.service.EsProcessInstanceService;
import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.flowable.engine.HistoryService;
import org.flowable.engine.history.HistoricProcessInstance;
import org.flowable.engine.impl.cfg.ProcessEngineConfigurationImpl;
import org.flowable.engine.impl.util.CommandContextUtil;
import org.flowable.task.service.impl.persistence.entity.TaskEntity;
import org.springframework.stereotype.Component;
/**
* 用于处理非结束的实例的数据同步至 ES
*
* @author wangli
* @since 2024-09-06 00:02
*/
@Slf4j
@Component
@AllArgsConstructor
public class SyncEsTaskEntityEventHandle implements EntityEventHandle<TaskEntity> {
private final AggregateProcessInstanceService aggregateProcessInstanceService;
private final EsProcessInstanceService esProcessInstanceService;
@Override
public boolean support(Object entity) {
return entity instanceof TaskEntity;
}
@Override
public TaskEntity convert(Object entity) {
return (TaskEntity) entity;
}
@Override
public void onInitialized(TaskEntity taskEntity) {
log.info("SyncEsTaskEntityEventHandle onInitialized processInstanceId:{}, taskId: {}", taskEntity.getProcessInstanceId(), taskEntity.getId());
esProcessInstanceService.delete(taskEntity.getProcessInstanceId());
log.info("delete document processInstanceId: {}", taskEntity.getProcessInstanceId());
ProcessEngineConfigurationImpl processEngineConfiguration = CommandContextUtil.getProcessEngineConfiguration();
HistoryService historyService = processEngineConfiguration.getHistoryService();
HistoricProcessInstance historicProcessInstance = historyService.createHistoricProcessInstanceQuery()
.processInstanceId(taskEntity.getProcessInstanceId())
.singleResult();
aggregateProcessInstanceService.syncProcessInstance(historicProcessInstance, null);
log.info("reInsert document processInstanceId: {}", taskEntity.getProcessInstanceId());
}
}

2
workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/task/SnapshotBpmnTaskTaskEvent_100_Listener.java → workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/task/SnapshotBpmnTaskEvent_100_Listener.java
View File

@ -34,7 +34,7 @@ import static cn.axzo.workflow.common.constant.BpmnConstants.INTERNAL_TASK_RELAT
@Component
@Scope("prototype")
@AllArgsConstructor
public class SnapshotBpmnTaskTaskEvent_100_Listener extends AbstractBpmnEventListener<TaskOperationContext> implements BpmnTaskEventListener, Ordered {
public class SnapshotBpmnTaskEvent_100_Listener extends AbstractBpmnEventListener<TaskOperationContext> implements BpmnTaskEventListener, Ordered {
@Override
public int getOrder() {
return Integer.MIN_VALUE + 100;

65
workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/task/SyncToEsTaskEvent_104_Listener.java Normal file
View File

@ -0,0 +1,65 @@
package cn.axzo.workflow.server.controller.listener.task;
import cn.axzo.workflow.core.common.context.TaskOperationContext;
import cn.axzo.workflow.core.listener.AbstractBpmnEventListener;
import cn.axzo.workflow.core.listener.BpmnTaskEventListener;
import cn.axzo.workflow.es.service.EsProcessInstanceService;
import cn.axzo.workflow.es.service.EsProcessTaskService;
import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
import cn.axzo.workflow.es.service.impl.EsProcessTaskServiceImpl;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.flowable.engine.HistoryService;
import org.flowable.engine.history.HistoricProcessInstance;
import org.flowable.engine.impl.cfg.ProcessEngineConfigurationImpl;
import org.flowable.engine.impl.util.CommandContextUtil;
import org.flowable.task.api.history.HistoricTaskInstance;
import org.flowable.task.service.delegate.DelegateTask;
import org.springframework.context.annotation.Scope;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 同步任务相关数据至 ES
*
* @author wangli
* @since 2024-10-16 16:50
*/
@Slf4j
@Component
@Scope("prototype")
@AllArgsConstructor
public class SyncToEsTaskEvent_104_Listener extends AbstractBpmnEventListener<TaskOperationContext> implements BpmnTaskEventListener, Ordered {
private final AggregateProcessInstanceService aggregateProcessInstanceService;
@Override
public int getOrder() {
return Integer.MIN_VALUE + 104;
}
/**
* 用户任务已指派审核人
*
* @param delegateTask
*/
@Override
public void onAssigned(DelegateTask delegateTask) {
log.info("SyncEsTaskEntityEventHandle onInitialized processInstanceId:{}, taskId: {}", delegateTask.getProcessInstanceId(), delegateTask.getId());
// 删除指定实例的父子文档
aggregateProcessInstanceService.deleteDocumentParentAndChild(delegateTask.getProcessInstanceId());
log.info("delete document processInstanceId: {}", delegateTask.getProcessInstanceId());
ProcessEngineConfigurationImpl processEngineConfiguration = CommandContextUtil.getProcessEngineConfiguration();
HistoryService historyService = processEngineConfiguration.getHistoryService();
HistoricProcessInstance historicProcessInstance = historyService.createHistoricProcessInstanceQuery()
.processInstanceId(delegateTask.getProcessInstanceId())
.singleResult();
aggregateProcessInstanceService.syncProcessInstance(historicProcessInstance, null);
log.info("reInsert document processInstanceId: {}", delegateTask.getProcessInstanceId());
}
}