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

REQ-2596-调整抄送mq消息按人员分批次拆分代码

Browse Source
This commit is contained in:
yangqicheng 2024-07-05 09:44:47 +08:00
parent b3b1be05de
commit a486798683
2 changed files with 18 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/listener/EngineCarbonCopyEventListener.java
View File

@ -108,17 +108,10 @@ public class EngineCarbonCopyEventListener implements JavaDelegate {
}
return conf;
}).orElse(null);
//按人员批次发送消息
int startIndex = 0, endIndex = Integer.min(startIndex + CARBON_ASSIGNER_BATCH_SIZE, carbonUsers.size());
do {
List<BpmnTaskDelegateAssigner> bpmnTaskDelegateAssigners = carbonUsers.subList(startIndex, endIndex);
MessagePushEventImpl event = MessagePushEventBuilder.createEvent(MessagePushEventType.CARBON_COPY,
bpmnTaskDelegateAssigners, bpmnNoticeConf, execution.getProcessInstanceId(),
execution.getTenantId(), getCarbonTaskId(execution));
eventDispatcher.dispatchEvent(event, processEngineConfiguration.getEngineCfgKey());
startIndex = endIndex;
endIndex = Integer.min(endIndex + CARBON_ASSIGNER_BATCH_SIZE, carbonUsers.size());
} while (startIndex < carbonUsers.size());
MessagePushEventImpl event = MessagePushEventBuilder.createEvent(MessagePushEventType.CARBON_COPY,
carbonUsers, bpmnNoticeConf, execution.getProcessInstanceId(),
execution.getTenantId(), getCarbonTaskId(execution));
eventDispatcher.dispatchEvent(event, processEngineConfiguration.getEngineCfgKey());
}
private String getCarbonTaskId(DelegateExecution execution) {

17
workflow-engine-server/src/main/java/cn/axzo/workflow/server/controller/listener/notice/RocketMqMessagePushEventListener.java
View File

@ -1,5 +1,6 @@
package cn.axzo.workflow.server.controller.listener.notice;
import cn.axzo.core.utils.converter.BeanConverter;
import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventProducer;
import cn.axzo.workflow.common.enums.ProcessMessagePushEventEnum;
@ -12,6 +13,7 @@ import cn.axzo.workflow.core.common.context.NoticeOperationContext;
import cn.axzo.workflow.core.common.utils.BpmnMetaParserHelper;
import cn.axzo.workflow.core.engine.cmd.helper.CustomTaskHelper;
import cn.axzo.workflow.core.engine.event.MessagePushEvent;
import cn.axzo.workflow.core.engine.event.MessagePushEventImpl;
import cn.axzo.workflow.core.listener.AbstractBpmnEventListener;
import cn.axzo.workflow.core.listener.BpmnMessagePushEventListener;
import cn.axzo.workflow.core.service.BpmnProcessInstanceService;
@ -43,6 +45,7 @@ import java.util.Objects;
import java.util.stream.Collectors;
import static cn.axzo.workflow.common.constant.BpmnConstants.BIZ_ORG_RELATION;
import static cn.axzo.workflow.common.constant.BpmnConstants.CARBON_ASSIGNER_BATCH_SIZE;
import static cn.axzo.workflow.common.constant.BpmnConstants.INTERNAL_ACTIVITY_RELATION_ASSIGNEE_LIST_INFO_SNAPSHOT;
import static cn.axzo.workflow.common.constant.BpmnConstants.INTERNAL_DELETE_PROCESS_FLAG;
import static cn.axzo.workflow.common.constant.BpmnConstants.INTERNAL_END_TENANT_ID;
@ -248,9 +251,17 @@ public class RocketMqMessagePushEventListener extends AbstractBpmnEventListener<
event.getNoticeConfig().getCarbonCopy().getCarbonCopyMessageId(),
JSONUtil.toJsonStr(event.getAssigners()), event.getProcessInstanceId());
if (Objects.nonNull(event.getNoticeConfig().getCarbonCopy())) {
MessagePushDTO dto = build(event.getNoticeConfig().getCarbonCopy().getCarbonCopyMessageId(),
PROCESS_CARBON_COPY, event, collectionVariable(event));
sendMessageQueue(dto, PROCESS_CARBON_COPY);
List<BpmnTaskDelegateAssigner> carbonAssigners = event.getAssigners();
//按人员拆分为多个批次发送消息
int startIndex = 0;
do {
List<BpmnTaskDelegateAssigner> batchAssigners = carbonAssigners.subList(startIndex, Integer.min(startIndex + CARBON_ASSIGNER_BATCH_SIZE, carbonAssigners.size()));
MessagePushEventImpl messagePushEvent = BeanConverter.convert(event, MessagePushEventImpl.class);
messagePushEvent.setAssigner(batchAssigners);
MessagePushDTO dto = build(event.getNoticeConfig().getCarbonCopy().getCarbonCopyMessageId(), PROCESS_CARBON_COPY, messagePushEvent, collectionVariable(event));
sendMessageQueue(dto, PROCESS_CARBON_COPY);
startIndex = startIndex + CARBON_ASSIGNER_BATCH_SIZE;
} while (startIndex < carbonAssigners.size());
}
log.info("RocketMqMessagePushEventListener#onCarbonCopy...end, cc' templateId: {}, receivePerson: {}, processInstanceId: {}",
event.getNoticeConfig().getCarbonCopy().getCarbonCopyMessageId(),