From d7a53ae3e6cf8913e0edf5f930f80242b51035ad Mon Sep 17 00:00:00 2001 From: byteblogs168 <598092184@qq.com> Date: Mon, 1 Jan 2024 23:20:32 +0800 Subject: [PATCH] =?UTF-8?q?feat:=202.6.0=201.=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E6=9D=A1=E4=BB=B6=E8=8A=82=E7=82=B9=E5=88=A4=E5=AE=9A=202.=20?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=9B=9E=E8=B0=83=E8=8A=82=E7=82=B9=E7=9A=84?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../server/common/dto/CallbackConfig.java | 2 +- .../common/enums/LogicalConditionEnum.java | 22 ++++++++ .../task/dto/WorkflowNodeTaskExecuteDTO.java | 5 +- .../support/dispatch/JobExecutorActor.java | 2 +- .../dispatch/WorkflowExecutorActor.java | 19 ++++--- .../workflow/CallbackWorkflowExecutor.java | 35 +++++++++++++ .../workflow/ConditionWorkflowExecutor.java | 52 +++++++++++-------- .../workflow/JobTaskWorkflowExecutor.java | 2 +- .../workflow/WorkflowExecutorContext.java | 18 ++++--- .../batch/JobTaskBatchGenerator.java | 1 - .../support/handler/JobTaskBatchHandler.java | 2 +- .../response/WorkflowDetailResponseVO.java | 2 +- .../convert/JobBatchResponseVOConverter.java | 2 + .../impl/WorkflowBatchServiceImpl.java | 13 +++-- 14 files changed, 126 insertions(+), 51 deletions(-) create mode 100644 easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/enums/LogicalConditionEnum.java diff --git a/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/dto/CallbackConfig.java b/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/dto/CallbackConfig.java index b5c10268b..62ee06064 100644 --- a/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/dto/CallbackConfig.java +++ b/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/dto/CallbackConfig.java @@ -15,7 +15,7 @@ public class CallbackConfig { /** * webhook */ - private Integer webhook; + private String webhook; /** * 请求类型 diff --git a/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/enums/LogicalConditionEnum.java b/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/enums/LogicalConditionEnum.java new file mode 100644 index 000000000..ebc71ed12 --- /dev/null +++ b/easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/enums/LogicalConditionEnum.java @@ -0,0 +1,22 @@ +package com.aizuda.easy.retry.server.common.enums; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +/** + * @author xiaowoniu + * @date 2024-01-01 22:56:28 + * @since 2.6.0 + */ +@Getter +@AllArgsConstructor +public enum LogicalConditionEnum { + /** + * 逻辑条件 + */ + AND(1, "并"), + OR(2, "或"); + + private final Integer code; + private final String desc; +} diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/dto/WorkflowNodeTaskExecuteDTO.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/dto/WorkflowNodeTaskExecuteDTO.java index 936b56737..3c8d79aa3 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/dto/WorkflowNodeTaskExecuteDTO.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/dto/WorkflowNodeTaskExecuteDTO.java @@ -26,5 +26,8 @@ public class WorkflowNodeTaskExecuteDTO { private Long parentId; - private String result; + /** + * 调度任务id + */ + private Long taskBatchId; } diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorActor.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorActor.java index 5dac34399..ace339c6c 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorActor.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorActor.java @@ -119,7 +119,7 @@ public class JobExecutorActor extends AbstractActor { taskExecuteDTO.setWorkflowTaskBatchId(taskExecute.getWorkflowTaskBatchId()); taskExecuteDTO.setTriggerType(JobTriggerTypeEnum.AUTO.getType()); taskExecuteDTO.setParentId(taskExecute.getWorkflowNodeId()); - taskExecuteDTO.setResult(StrUtil.EMPTY); + taskExecuteDTO.setTaskBatchId(taskExecute.getTaskBatchId()); ActorRef actorRef = ActorGenerator.workflowTaskExecutorActor(); actorRef.tell(taskExecuteDTO, actorRef); } catch (Exception e) { diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/WorkflowExecutorActor.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/WorkflowExecutorActor.java index 0b631ddc5..576d09629 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/WorkflowExecutorActor.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/WorkflowExecutorActor.java @@ -100,7 +100,7 @@ public class WorkflowExecutorActor extends AbstractActor { } // 添加父节点，为了判断父节点的处理状态 - List jobTaskBatchList = jobTaskBatchMapper.selectList(new LambdaQueryWrapper() + List allJobTaskBatchList = jobTaskBatchMapper.selectList(new LambdaQueryWrapper() .select(JobTaskBatch::getWorkflowTaskBatchId, JobTaskBatch::getWorkflowNodeId, JobTaskBatch::getTaskBatchStatus) .eq(JobTaskBatch::getWorkflowTaskBatchId, workflowTaskBatch.getId()) .in(JobTaskBatch::getWorkflowNodeId, Sets.union(successors, Sets.newHashSet(taskExecute.getParentId()))) @@ -109,12 +109,15 @@ public class WorkflowExecutorActor extends AbstractActor { List workflowNodes = workflowNodeMapper.selectList(new LambdaQueryWrapper() .in(WorkflowNode::getId, Sets.union(successors, Sets.newHashSet(taskExecute.getParentId()))).orderByAsc(WorkflowNode::getPriorityLevel)); - Map jobTaskBatchMap = jobTaskBatchList.stream().collect(Collectors.toMap(JobTaskBatch::getWorkflowNodeId, i -> i)); + Map> jobTaskBatchMap = allJobTaskBatchList.stream().collect(Collectors.groupingBy(JobTaskBatch::getWorkflowNodeId)); Map workflowNodeMap = workflowNodes.stream().collect(Collectors.toMap(WorkflowNode::getId, i -> i)); - JobTaskBatch parentJobTaskBatch = jobTaskBatchMap.get(taskExecute.getParentId()); + List parentJobTaskBatchList = jobTaskBatchMap.get(taskExecute.getParentId()); // 失败策略处理 - if (Objects.nonNull(parentJobTaskBatch) && JobTaskBatchStatusEnum.SUCCESS.getStatus() != parentJobTaskBatch.getTaskBatchStatus()) { + if (!CollectionUtils.isEmpty(parentJobTaskBatchList) + && parentJobTaskBatchList.stream() + .map(JobTaskBatch::getTaskBatchStatus) + .anyMatch(i -> i != JobTaskBatchStatusEnum.SUCCESS.getStatus())) { // 判断是否继续处理，根据失败策略 WorkflowNode workflowNode = workflowNodeMap.get(taskExecute.getParentId()); // 失败了阻塞策略 @@ -125,7 +128,7 @@ public class WorkflowExecutorActor extends AbstractActor { // 去掉父节点 workflowNodes = workflowNodes.stream().filter(workflowNode -> !workflowNode.getId().equals(taskExecute.getParentId())).collect( - Collectors.toList()); + Collectors.toList()); List jobs = jobMapper.selectBatchIds(workflowNodes.stream().map(WorkflowNode::getJobId).collect(Collectors.toSet())); Map jobMap = jobs.stream().collect(Collectors.toMap(Job::getId, i -> i)); @@ -135,8 +138,8 @@ public class WorkflowExecutorActor extends AbstractActor { for (WorkflowNode workflowNode : workflowNodes) { // 批次已经存在就不在重复生成 - JobTaskBatch jobTaskBatch = jobTaskBatchMap.get(workflowNode.getId()); - if (Objects.nonNull(jobTaskBatch) && JobTaskBatchStatusEnum.COMPLETED.contains(jobTaskBatch.getTaskBatchStatus())) { + List jobTaskBatchList = jobTaskBatchMap.get(workflowNode.getId()); + if (!CollectionUtils.isEmpty(jobTaskBatchList)) { continue; } @@ -147,8 +150,8 @@ public class WorkflowExecutorActor extends AbstractActor { context.setJob(jobMap.get(workflowNode.getJobId())); context.setWorkflowTaskBatchId(taskExecute.getWorkflowTaskBatchId()); context.setParentWorkflowNodeId(taskExecute.getParentId()); - context.setResult(taskExecute.getResult()); context.setEvaluationResult(evaluationResult); + context.setTaskBatchId(taskExecute.getTaskBatchId()); workflowExecutor.execute(context); diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java index 5d93e7711..93141fb15 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java @@ -1,7 +1,21 @@ package com.aizuda.easy.retry.server.job.task.support.executor.workflow; import com.aizuda.easy.retry.common.core.enums.WorkflowNodeTypeEnum; +import com.aizuda.easy.retry.common.core.util.JsonUtil; +import com.aizuda.easy.retry.server.common.dto.CallbackConfig; +import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobTaskMapper; +import com.aizuda.easy.retry.template.datasource.persistence.po.JobTask; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import lombok.RequiredArgsConstructor; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; import org.springframework.stereotype.Component; +import org.springframework.web.client.RestTemplate; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; /** * @author xiaowoniu @@ -9,7 +23,10 @@ import org.springframework.stereotype.Component; * @since 2.6.0 */ @Component +@RequiredArgsConstructor public class CallbackWorkflowExecutor extends AbstractWorkflowExecutor { + private final RestTemplate restTemplate; + private final JobTaskMapper jobTaskMapper; @Override public WorkflowNodeTypeEnum getWorkflowNodeType() { return WorkflowNodeTypeEnum.CALLBACK; @@ -17,6 +34,24 @@ public class CallbackWorkflowExecutor extends AbstractWorkflowExecutor { @Override protected void doExecute(WorkflowExecutorContext context) { + CallbackConfig decisionConfig = JsonUtil.parseObject(context.getNodeInfo(), CallbackConfig.class); + HttpHeaders requestHeaders = new HttpHeaders(); + requestHeaders.set(HttpHeaders.CONTENT_TYPE, decisionConfig.getContentType()); + requestHeaders.set("secret", decisionConfig.getSecret()); + + // TODO 拼接所有的任务结果值传递给下游节点 + List jobTasks = jobTaskMapper.selectList(new LambdaQueryWrapper() + .select(JobTask::getResultMessage) + .eq(JobTask::getTaskBatchId, context.getTaskBatchId())); + + Map uriVariables = new HashMap<>(); + uriVariables.put("secret", decisionConfig.getSecret()); + restTemplate.exchange(decisionConfig.getWebhook(), HttpMethod.POST, + new HttpEntity<>("", requestHeaders), Object.class, uriVariables); + + // TODO 保存批次 + // TODO 保存任务 + // TODO 保存日志 } } diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/ConditionWorkflowExecutor.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/ConditionWorkflowExecutor.java index a0b33b13c..265ff76fa 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/ConditionWorkflowExecutor.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/ConditionWorkflowExecutor.java @@ -12,8 +12,10 @@ import com.aizuda.easy.retry.common.core.expression.ExpressionEngine; import com.aizuda.easy.retry.common.core.expression.ExpressionFactory; import com.aizuda.easy.retry.common.core.util.JsonUtil; import com.aizuda.easy.retry.server.common.akka.ActorGenerator; +import com.aizuda.easy.retry.server.common.dto.DecisionConfig; import com.aizuda.easy.retry.server.common.enums.ExpressionTypeEnum; import com.aizuda.easy.retry.server.common.enums.JobTriggerTypeEnum; +import com.aizuda.easy.retry.server.common.enums.LogicalConditionEnum; import com.aizuda.easy.retry.server.common.exception.EasyRetryServerException; import com.aizuda.easy.retry.server.job.task.dto.JobLogDTO; import com.aizuda.easy.retry.server.job.task.dto.WorkflowNodeTaskExecuteDTO; @@ -24,6 +26,7 @@ import com.aizuda.easy.retry.server.job.task.support.generator.batch.JobTaskBatc import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobTaskMapper; import com.aizuda.easy.retry.template.datasource.persistence.po.JobTask; import com.aizuda.easy.retry.template.datasource.persistence.po.JobTaskBatch; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.expression.EvaluationContext; @@ -32,9 +35,7 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.stereotype.Component; -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; +import java.util.*; /** * @author xiaowoniu @@ -45,9 +46,6 @@ import java.util.Optional; @RequiredArgsConstructor @Slf4j public class ConditionWorkflowExecutor extends AbstractWorkflowExecutor { - - private static final ExpressionParser ENGINE = new SpelExpressionParser(); - private final JobTaskBatchGenerator jobTaskBatchGenerator; private final JobTaskMapper jobTaskMapper; @@ -69,15 +67,36 @@ public class ConditionWorkflowExecutor extends AbstractWorkflowExecutor { // 多个条件节点直接是或的关系，只要一个成功其他节点就取消 taskBatchStatus = JobTaskBatchStatusEnum.CANCEL.getStatus(); } else { + DecisionConfig decisionConfig = JsonUtil.parseObject(context.getNodeInfo(), DecisionConfig.class); try { - ExpressionEngine realExpressionEngine = ExpressionTypeEnum.valueOf(context.getExpressionType()); + ExpressionEngine realExpressionEngine = ExpressionTypeEnum.valueOf(decisionConfig.getExpressionType()); Assert.notNull(realExpressionEngine, () -> new EasyRetryServerException("表达式引擎不存在")); ExpressionInvocationHandler invocationHandler = new ExpressionInvocationHandler(realExpressionEngine); ExpressionEngine expressionEngine = ExpressionFactory.getExpressionEngine(invocationHandler); - result = (Boolean) Optional.ofNullable(expressionEngine.eval(context.getNodeExpression(), context.getResult())).orElse(Boolean.FALSE); - log.info("执行条件表达式：[{}]，参数: [{}] 结果：[{}]", context.getNodeExpression(), context.getResult(), result); + + List jobTasks = jobTaskMapper.selectList(new LambdaQueryWrapper() + .select(JobTask::getResultMessage) + .eq(JobTask::getTaskBatchId, context.getTaskBatchId())); + Boolean tempResult = Boolean.TRUE; + for (JobTask jobTask : jobTasks) { + Boolean execResult = (Boolean) Optional.ofNullable(expressionEngine.eval(decisionConfig.getNodeExpression(), jobTask.getResultMessage())).orElse(Boolean.FALSE); + + if (Objects.equals(decisionConfig.getLogicalCondition(), LogicalConditionEnum.AND.getCode())) { + tempResult = tempResult && execResult; + } else { + tempResult = tempResult || execResult; + if (tempResult) { + break; + } + } + + log.info("执行条件表达式：[{}]，参数: [{}] 结果：[{}]", decisionConfig.getNodeExpression(), jobTask.getResultMessage(), result); + + } + + result = tempResult; } catch (Exception e) { - log.error("执行条件表达式解析异常. 表达式:[{}]，参数: [{}]", context.getNodeExpression(), context.getResult(), e); + log.error("执行条件表达式解析异常. 表达式:[{}]，参数: [{}]", decisionConfig.getNodeExpression(), context.getResult(), e); taskBatchStatus = JobTaskBatchStatusEnum.FAIL.getStatus(); operationReason = JobOperationReasonEnum.WORKFLOW_CONDITION_NODE_EXECUTOR_ERROR.getReason(); jobTaskStatus = JobTaskStatusEnum.FAIL.getStatus(); @@ -92,7 +111,7 @@ public class ConditionWorkflowExecutor extends AbstractWorkflowExecutor { taskExecuteDTO.setWorkflowTaskBatchId(context.getWorkflowTaskBatchId()); taskExecuteDTO.setTriggerType(JobTriggerTypeEnum.AUTO.getType()); taskExecuteDTO.setParentId(context.getWorkflowNodeId()); - taskExecuteDTO.setResult(context.getResult()); + taskExecuteDTO.setTaskBatchId(context.getTaskBatchId()); ActorRef actorRef = ActorGenerator.workflowTaskExecutorActor(); actorRef.tell(taskExecuteDTO, actorRef); } catch (Exception e) { @@ -135,15 +154,4 @@ public class ConditionWorkflowExecutor extends AbstractWorkflowExecutor { actorRef.tell(jobLogDTO, actorRef); } - protected Boolean doEval(String expression, Map context) { - try { - final EvaluationContext evaluationContext = new StandardEvaluationContext(); - context.forEach(evaluationContext::setVariable); - return ENGINE.parseExpression(expression).getValue(evaluationContext, Boolean.class); - } catch (Exception e) { - throw new EasyRetryServerException("SpEL表达式解析异常. expression:[{}] context:[{}]", - expression, JsonUtil.toJsonString(context), e); - } - - } } diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/JobTaskWorkflowExecutor.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/JobTaskWorkflowExecutor.java index 289557ca5..6c8119359 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/JobTaskWorkflowExecutor.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/JobTaskWorkflowExecutor.java @@ -27,7 +27,7 @@ public class JobTaskWorkflowExecutor extends AbstractWorkflowExecutor { // 生成任务批次 JobTaskPrepareDTO jobTaskPrepare = JobTaskConverter.INSTANCE.toJobTaskPrepare(context.getJob()); jobTaskPrepare.setTriggerType(JobTriggerTypeEnum.WORKFLOW.getType()); - jobTaskPrepare.setNextTriggerAt(DateUtils.toNowMilli()); + jobTaskPrepare.setNextTriggerAt(DateUtils.toNowMilli() + DateUtils.toNowMilli() % 1000); jobTaskPrepare.setWorkflowNodeId(context.getWorkflowNodeId()); jobTaskPrepare.setWorkflowTaskBatchId(context.getWorkflowTaskBatchId()); jobTaskPrepare.setParentWorkflowNodeId(context.getParentWorkflowNodeId()); diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/WorkflowExecutorContext.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/WorkflowExecutorContext.java index 3ee5a1f94..d588f2d4c 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/WorkflowExecutorContext.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/WorkflowExecutorContext.java @@ -43,18 +43,11 @@ public class WorkflowExecutorContext { */ private Job job; - private String nodeExpression; - /** * 客户端返回的结果 */ private String result; - /** - * 1、SpEl、2、Aviator 3、QL - */ - private Integer expressionType; - /** * 失败策略 1、跳过 2、阻塞 */ @@ -70,4 +63,15 @@ public class WorkflowExecutorContext { */ private Boolean evaluationResult; + /** + * 调度任务id + */ + private Long taskBatchId; + + /** + * 节点信息 + */ + private String nodeInfo; + + } diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/generator/batch/JobTaskBatchGenerator.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/generator/batch/JobTaskBatchGenerator.java index 5f4fad130..fd190e0cc 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/generator/batch/JobTaskBatchGenerator.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/generator/batch/JobTaskBatchGenerator.java @@ -67,7 +67,6 @@ public class JobTaskBatchGenerator { taskExecuteDTO.setWorkflowTaskBatchId(context.getWorkflowTaskBatchId()); taskExecuteDTO.setTriggerType(JobTriggerTypeEnum.AUTO.getType()); taskExecuteDTO.setParentId(context.getWorkflowNodeId()); - taskExecuteDTO.setResult(StrUtil.EMPTY); ActorRef actorRef = ActorGenerator.workflowTaskExecutorActor(); actorRef.tell(taskExecuteDTO, actorRef); } catch (Exception e) { diff --git a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/JobTaskBatchHandler.java b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/JobTaskBatchHandler.java index 648b5f68c..87d370d3f 100644 --- a/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/JobTaskBatchHandler.java +++ b/easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/JobTaskBatchHandler.java @@ -86,7 +86,7 @@ public class JobTaskBatchHandler { taskExecuteDTO.setTriggerType(JobTriggerTypeEnum.AUTO.getType()); taskExecuteDTO.setParentId(completeJobBatchDTO.getWorkflowNodeId()); // 这里取第一个的任务执行结果 - taskExecuteDTO.setResult(jobTasks.get(0).getResultMessage()); + taskExecuteDTO.setTaskBatchId(completeJobBatchDTO.getTaskBatchId()); ActorRef actorRef = ActorGenerator.workflowTaskExecutorActor(); actorRef.tell(taskExecuteDTO, actorRef); } catch (Exception e) { diff --git a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/model/response/WorkflowDetailResponseVO.java b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/model/response/WorkflowDetailResponseVO.java index 9a90f5fdb..6854781ee 100644 --- a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/model/response/WorkflowDetailResponseVO.java +++ b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/model/response/WorkflowDetailResponseVO.java @@ -127,7 +127,7 @@ public class WorkflowDetailResponseVO { /** * 定时任务批次信息 */ - private JobBatchResponseVO jobBatch; + private List jobBatchList; /** * 子节点 diff --git a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/convert/JobBatchResponseVOConverter.java b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/convert/JobBatchResponseVOConverter.java index 560a1096c..f9411ad0d 100644 --- a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/convert/JobBatchResponseVOConverter.java +++ b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/convert/JobBatchResponseVOConverter.java @@ -31,6 +31,8 @@ public interface JobBatchResponseVOConverter { }) JobBatchResponseVO toJobBatchResponseVO(JobBatchResponseDO jobBatchResponseDO); + List jobTaskBatchToJobBatchResponseVOs(List jobTaskBatchList); + @Mappings({ @Mapping(target = "executionAt", expression = "java(JobBatchResponseVOConverter.toLocalDateTime(jobTaskBatch.getExecutionAt()))") }) diff --git a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/WorkflowBatchServiceImpl.java b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/WorkflowBatchServiceImpl.java index 1a94c3ea2..0ff6ce0c7 100644 --- a/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/WorkflowBatchServiceImpl.java +++ b/easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/WorkflowBatchServiceImpl.java @@ -107,19 +107,18 @@ public class WorkflowBatchServiceImpl implements WorkflowBatchService { .eq(WorkflowNode::getDeleted, StatusEnum.NO.getStatus()) .eq(WorkflowNode::getWorkflowId, workflow.getId())); - List jobTaskBatchList = jobTaskBatchMapper.selectList(new LambdaQueryWrapper() + List alJobTaskBatchList = jobTaskBatchMapper.selectList(new LambdaQueryWrapper() .eq(JobTaskBatch::getWorkflowTaskBatchId, id)); - Map jobTaskBatchMap = jobTaskBatchList.stream() - .collect(Collectors.toMap(JobTaskBatch::getWorkflowNodeId, i -> i, (v1, v2) -> v1)); + Map> jobTaskBatchMap = alJobTaskBatchList.stream() + .collect(Collectors.groupingBy(JobTaskBatch::getWorkflowNodeId)); List nodeInfos = WorkflowConverter.INSTANCE.toNodeInfo(workflowNodes); Map workflowNodeMap = nodeInfos.stream() .peek(nodeInfo -> { - JobTaskBatch jobTaskBatch = jobTaskBatchMap.get(nodeInfo.getId()); - if (Objects.nonNull(jobTaskBatch)) { - JobBatchResponseVO jobBatchResponseVO = JobBatchResponseVOConverter.INSTANCE.toJobBatchResponseVO(jobTaskBatch); - nodeInfo.setJobBatch(jobBatchResponseVO); + List jobTaskBatchList = jobTaskBatchMap.get(nodeInfo.getId()); + if (!CollectionUtils.isEmpty(jobTaskBatchList)) { + nodeInfo.setJobBatchList(JobBatchResponseVOConverter.INSTANCE.jobTaskBatchToJobBatchResponseVOs(jobTaskBatchList)); } }) .collect(Collectors.toMap(WorkflowDetailResponseVO.NodeInfo::getId, i -> i));