【功能新增】新增异步 http 请求触发器

jason
1 parent b88c09f4
Showing 11 changed files with 211 additions and 54 deletions
yudao-module-bpm/yudao-module-bpm-api/src/main/java/cn/iocoder/yudao/module/bpm/api/task/BpmProcessInstanceApi.java
yudao-module-bpm/yudao-module-bpm-api/src/main/java/cn/iocoder/yudao/module/bpm/enums/definition/BpmTriggerTypeEnum.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/api/task/BpmProcessInstanceApiImpl.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/controller/admin/definition/vo/model/simple/BpmSimpleModelNodeVO.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/framework/flowable/core/listener/BpmTaskEventListener.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/framework/flowable/core/util/SimpleModelUtils.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/task/BpmTaskService.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/task/BpmTaskServiceImpl.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/task/trigger/BpmAbstractHttpRequestTrigger.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/task/trigger/BpmAsyncHttpRequestTrigger.java
@@ -20,4 +20,12 @@ public interface BpmProcessInstanceApi {
      */
     String createProcessInstance(Long userId, @Valid BpmProcessInstanceCreateReqDTO reqDTO);
+
+    /**
+     * 异步 HTTP 请求触发器回调, 为了唤醒流程继续执行
+     *
+     * @param processInstanceId 流程实例编号
+     * @param callbackId 回调编号, 对应 ReceiveTask Id
+     */
+    void asyncHttpTriggerCallback(String processInstanceId, String callbackId);
 }
@@ -18,7 +18,8 @@ public enum BpmTriggerTypeEnum implements ArrayValuable&lt;Integer&gt; {
     HTTP_REQUEST(1, "发起 HTTP 请求"),
     FORM_UPDATE(2, "更新流程表单数据"),
-    FORM_DELETE(3, "删除流程表单数据");
+    FORM_DELETE(3, "删除流程表单数据"),
+    ASYNC_HTTP_REQUEST(4, "发起异步 HTTP 请求");
     /**
      * 触发器执行动作类型
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.bpm.api.task;
 import cn.iocoder.yudao.module.bpm.api.task.dto.BpmProcessInstanceCreateReqDTO;
 import cn.iocoder.yudao.module.bpm.service.task.BpmProcessInstanceService;
+import cn.iocoder.yudao.module.bpm.service.task.BpmTaskService;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
@@ -21,8 +22,16 @@ public class BpmProcessInstanceApiImpl implements BpmProcessInstanceApi {
     @Resource
     private BpmProcessInstanceService processInstanceService;
+    @Resource
+    private BpmTaskService bpmTaskService;
+
     @Override
     public String createProcessInstance(Long userId, @Valid BpmProcessInstanceCreateReqDTO reqDTO) {
         return processInstanceService.createProcessInstance(userId, reqDTO);
     }
+
+    @Override
+    public void asyncHttpTriggerCallback(String processInstanceId, String callbackId) {
+        bpmTaskService.triggerReceiveTask(processInstanceId, callbackId);
+    }
 }
@@ -4,6 +4,7 @@ import cn.iocoder.yudao.framework.common.core.KeyValue;
 import cn.iocoder.yudao.framework.common.validation.InEnum;
 import cn.iocoder.yudao.module.bpm.enums.definition.*;
 import cn.iocoder.yudao.module.bpm.framework.flowable.core.enums.BpmTaskCandidateStrategyEnum;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
@@ -123,6 +124,13 @@ public class BpmSimpleModelNodeVO {
      */
     private TriggerSetting triggerSetting;
+    /**
+     * 附加节点 Id, 该节点不从前端传入。 由程序生成. 由于当个节点无法完成功能。 需要附加节点来完成。
+     */
+    @JsonIgnore
+    private String attachNodeId; // 目前用于异步触发器节点。需要 UserTask 和 ReceiveTask（附加节点) 来完成
+
+
     @Schema(description = "任务监听器")
     @Valid
     @Data
@@ -378,6 +386,10 @@ public class BpmSimpleModelNodeVO {
             @Schema(description = "请求返回处理设置", example = "[]")
             private List<KeyValue<String, String>> response;
+            /**
+             * 异步 Http 请求，需要指定回调 Id，用于回调执行. 由后端指定
+             */
+            private String callbackId;
         }
         @Schema(description = "流程表单触发器设置", example = "{}")
@@ -109,8 +109,7 @@ public class BpmTaskEventListener extends AbstractFlowableEngineEventListener {
             // 2.2 延迟器超时处理
         } else if (ObjectUtil.equal(bpmTimerBoundaryEventType, BpmBoundaryEventTypeEnum.DELAY_TIMER_TIMEOUT)) {
             String taskKey = boundaryEvent.getAttachedToRefId();
-            taskService.processDelayTimerTimeout(event.getProcessInstanceId(), taskKey);
+            taskService.triggerReceiveTask(event.getProcessInstanceId(), taskKey);
         }
     }
-
 }
@@ -21,6 +21,7 @@ import org.springframework.util.MultiValueMap;
 import java.util.*;
+import static cn.iocoder.yudao.module.bpm.enums.definition.BpmTriggerTypeEnum.ASYNC_HTTP_REQUEST;
 import static cn.iocoder.yudao.module.bpm.framework.flowable.core.enums.BpmnModelConstants.*;
 import static cn.iocoder.yudao.module.bpm.framework.flowable.core.util.BpmnModelUtils.*;
 import static cn.iocoder.yudao.module.bpm.service.task.listener.BpmUserTaskListener.DELEGATE_EXPRESSION;
@@ -164,8 +165,15 @@ public class SimpleModelUtils {
         // 情况一：有“子”节点，则建立连线
         // 情况二：没有“子节点”，则直接跟 targetNodeId 建立连线。例如说，结束节点、条件分支（分支节点的孩子节点或聚合节点）的最后一个节点
         String finalTargetNodeId = isChildNodeValid ? childNode.getId() : targetNodeId;
-        SequenceFlow sequenceFlow = buildBpmnSequenceFlow(node.getId(), finalTargetNodeId);
-        process.addFlowElement(sequenceFlow);
+
+        if (StrUtil.isEmpty(node.getAttachNodeId())) {
+            SequenceFlow sequenceFlow = buildBpmnSequenceFlow(node.getId(), finalTargetNodeId);
+            process.addFlowElement(sequenceFlow);
+        } else {
+            // 如果有附加节点. 需要先建立和附加节点的连线。再建立附加节点和目标节点的连线
+            List<SequenceFlow> sequenceFlows = buildAttachNodeSequenceFlow(node.getId(), node.getAttachNodeId(), finalTargetNodeId);
+            sequenceFlows.forEach(process::addFlowElement);
+        }
         // 因为有子节点，递归调用后续子节点
         if (isChildNodeValid) {
@@ -174,6 +182,19 @@ public class SimpleModelUtils {
     }
     /**
+     * 构建有附加节点的连线
+     *
+     * @param nodeId 当前节点 Id
+     * @param attachNodeId 附属节点 Id
+     * @param targetNodeId 目标节点 Id
+     */
+    private static List<SequenceFlow> buildAttachNodeSequenceFlow(String nodeId, String attachNodeId, String targetNodeId) {
+        SequenceFlow sequenceFlow = buildBpmnSequenceFlow(nodeId, attachNodeId, null, null, null);
+        SequenceFlow attachSequenceFlow = buildBpmnSequenceFlow(attachNodeId, targetNodeId, null, null, null);
+        return CollUtil.newArrayList(sequenceFlow, attachSequenceFlow);
+    }
+
+    /**
      * 遍历条件节点，构建 SequenceFlow 元素
      *
      * @param process      Bpmn 流程
@@ -723,23 +744,38 @@ public class SimpleModelUtils {
     public static class TriggerNodeConvert implements NodeConvert {
         @Override
-        public ServiceTask convert(BpmSimpleModelNodeVO node) {
+        public List<? extends FlowElement> convertList(BpmSimpleModelNodeVO node) {
+            List<FlowElement> flowElements = new ArrayList<>(2);
             // 触发器使用 ServiceTask 来实现
             ServiceTask serviceTask = new ServiceTask();
             serviceTask.setId(node.getId());
             serviceTask.setName(node.getName());
             serviceTask.setImplementationType(ImplementationType.IMPLEMENTATION_TYPE_DELEGATEEXPRESSION);
             serviceTask.setImplementation("${" + BpmTriggerTaskDelegate.BEAN_NAME + "}");
-            if (node.getTriggerSetting() != null) {
-                addExtensionElement(serviceTask, TRIGGER_TYPE, node.getTriggerSetting().getType());
-                if (node.getTriggerSetting().getHttpRequestSetting() != null) {
-                    addExtensionElementJson(serviceTask, TRIGGER_PARAM, node.getTriggerSetting().getHttpRequestSetting());
-                }
-                if (node.getTriggerSetting().getFormSettings() != null) {
-                    addExtensionElementJson(serviceTask, TRIGGER_PARAM, node.getTriggerSetting().getFormSettings());
-                }
+            Assert.notNull(node.getTriggerSetting(), "触发器节点设置不能为空");
+
+            addExtensionElement(serviceTask, TRIGGER_TYPE, node.getTriggerSetting().getType());
+            flowElements.add(serviceTask);
+            // 异步 HTTP 请求。需要附加一个 ReceiveTask、发起请求后、等待回调执行
+            if (ASYNC_HTTP_REQUEST.getType().equals(node.getTriggerSetting().getType())) {
+                String attachNodeId = "Activity_" + IdUtil.fastUUID();
+                ReceiveTask receiveTask = new ReceiveTask();
+                receiveTask.setId(attachNodeId);
+                receiveTask.setName("异步 HTTP 请求");
+                node.setAttachNodeId(attachNodeId);
+                // 设置 receiveId
+                Assert.notNull(node.getTriggerSetting().getHttpRequestSetting(), "触发器 http 请求设置不能为空");
+                node.getTriggerSetting().getHttpRequestSetting().setCallbackId(attachNodeId);
+                flowElements.add(receiveTask);
             }
-            return serviceTask;
+
+            if (node.getTriggerSetting().getHttpRequestSetting() != null) {
+                addExtensionElementJson(serviceTask, TRIGGER_PARAM, node.getTriggerSetting().getHttpRequestSetting());
+            }
+            if (node.getTriggerSetting().getFormSettings() != null) {
+                addExtensionElementJson(serviceTask, TRIGGER_PARAM, node.getTriggerSetting().getFormSettings());
+            }
+            return flowElements;
         }
         @Override
@@ -277,11 +277,11 @@ public interface BpmTaskService {
     void processTaskTimeout(String processInstanceId, String taskDefineKey, Integer handlerType);
     /**
-     * 处理 延迟器 超时事件
+     * 触发 ReceiveTask, 让流程继续执行
      *
      * @param processInstanceId 流程示例编号
      * @param taskDefineKey     任务 Key
      */
-    void processDelayTimerTimeout(String processInstanceId, String taskDefineKey);
+    void triggerReceiveTask(String processInstanceId, String taskDefineKey);
 }
@@ -1309,7 +1309,7 @@ public class BpmTaskServiceImpl implements BpmTaskService {
     }
     @Override
-    public void processDelayTimerTimeout(String processInstanceId, String taskDefineKey) {
+    public void triggerReceiveTask(String processInstanceId, String taskDefineKey) {
         Execution execution = runtimeService.createExecutionQuery()
                 .processInstanceId(processInstanceId)
                 .activityId(taskDefineKey)
+package cn.iocoder.yudao.module.bpm.service.task.trigger;
+
+import cn.iocoder.yudao.module.bpm.controller.admin.definition.vo.model.simple.BpmSimpleModelNodeVO;
+import cn.iocoder.yudao.module.bpm.framework.flowable.core.util.SimpleModelUtils;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.flowable.engine.runtime.ProcessInstance;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.client.RestClientException;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.List;
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID;
+
+/**
+ * BPM 发送 HTTP 请求触发器抽象类
+ *
+ * @author jason
+ */
+@Slf4j
+public abstract class BpmAbstractHttpRequestTrigger implements BpmTrigger {
+
+    @Resource
+    private RestTemplate restTemplate;
+
+    protected ResponseEntity<String> sendHttpRequest(String url,
+                                                     MultiValueMap<String, String> headers,
+                                                     MultiValueMap<String, String> body) {
+        // TODO @芋艿：要不要抽象一个 Http 请求的工具类，方便复用呢？
+        // 3. 发起请求
+        HttpEntity<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(body, headers);
+        ResponseEntity<String> responseEntity = null;
+        try {
+            responseEntity = restTemplate.exchange(url, HttpMethod.POST,
+                    requestEntity, String.class);
+            log.info("[sendHttpRequest][HTTP 触发器，请求头：{}，请求体：{}，响应结果：{}]", headers, body, responseEntity);
+        } catch (RestClientException e) {
+            log.error("[sendHttpRequest][HTTP 触发器，请求头：{}，请求体：{}，请求出错：{}]", headers, body, e.getMessage());
+        }
+        return responseEntity;
+    }
+
+    protected MultiValueMap<String, String> buildHttpHeaders(ProcessInstance processInstance,
+                                                             List<BpmSimpleModelNodeVO.HttpRequestParam> headerSettings) {
+        Map<String, Object> processVariables = processInstance.getProcessVariables();
+        MultiValueMap<String, String> headers = new LinkedMultiValueMap<>();
+        headers.add(HEADER_TENANT_ID, processInstance.getTenantId());
+        SimpleModelUtils.addHttpRequestParam(headers, headerSettings, processVariables);
+        return headers;
+    }
+
+    protected MultiValueMap<String, String> buildHttpBody(ProcessInstance processInstance,
+                                                             List<BpmSimpleModelNodeVO.HttpRequestParam> bodySettings) {
+        Map<String, Object> processVariables = processInstance.getProcessVariables();
+        MultiValueMap<String, String> body = new LinkedMultiValueMap<>();
+        SimpleModelUtils.addHttpRequestParam(body, bodySettings, processVariables);
+        body.add("processInstanceId", processInstance.getId());
+        return body;
+    }
+}
+package cn.iocoder.yudao.module.bpm.service.task.trigger;
+
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.bpm.controller.admin.definition.vo.model.simple.BpmSimpleModelNodeVO;
+import cn.iocoder.yudao.module.bpm.enums.definition.BpmTriggerTypeEnum;
+import cn.iocoder.yudao.module.bpm.service.task.BpmProcessInstanceService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.flowable.engine.runtime.ProcessInstance;
+import org.springframework.stereotype.Component;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * BPM 发送异步 HTTP 请求触发器
+ *
+ * @author jason
+ */
+@Component
+@Slf4j
+public class BpmAsyncHttpRequestTrigger extends BpmAbstractHttpRequestTrigger {
+
+    @Resource
+    private BpmProcessInstanceService processInstanceService;
+
+    @Override
+    public BpmTriggerTypeEnum getType() {
+        return BpmTriggerTypeEnum.ASYNC_HTTP_REQUEST;
+    }
+
+    @Override
+    public void execute(String processInstanceId, String param) {
+        // 1. 解析 http 请求配置
+        BpmSimpleModelNodeVO.TriggerSetting.HttpRequestTriggerSetting setting = JsonUtils.parseObject(param,
+                BpmSimpleModelNodeVO.TriggerSetting.HttpRequestTriggerSetting.class);
+        if (setting == null) {
+            log.error("[execute][流程({}) HTTP 异步触发器请求配置为空]", processInstanceId);
+            return;
+        }
+
+        // 2.1 设置请求头
+        ProcessInstance processInstance = processInstanceService.getProcessInstance(processInstanceId);
+        MultiValueMap<String, String> headers = buildHttpHeaders(processInstance, setting.getHeader());
+        // 2.2 设置请求体
+        MultiValueMap<String, String> body = buildHttpBody(processInstance, setting.getBody());
+        body.add("callbackId", setting.getCallbackId()); // 异步请求 callbackId 需要传给被调用方. 用于回调执行
+
+        // 3. 发起请求
+        sendHttpRequest(setting.getUrl(), headers, body);
+    }
+}
@@ -7,42 +7,31 @@ import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.module.bpm.controller.admin.definition.vo.model.simple.BpmSimpleModelNodeVO.TriggerSetting.HttpRequestTriggerSetting;
 import cn.iocoder.yudao.module.bpm.enums.definition.BpmTriggerTypeEnum;
-import cn.iocoder.yudao.module.bpm.framework.flowable.core.util.SimpleModelUtils;
 import cn.iocoder.yudao.module.bpm.service.task.BpmProcessInstanceService;
 import com.fasterxml.jackson.core.type.TypeReference;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.engine.runtime.ProcessInstance;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
-import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
-import org.springframework.web.client.RestClientException;
-import org.springframework.web.client.RestTemplate;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID;
-
 /**
- * BPM 发送 HTTP 请求触发器
+ * BPM 发送同步 HTTP 请求触发器
  *
  * @author jason
  */
 @Component
 @Slf4j
-public class BpmHttpRequestTrigger implements BpmTrigger {
+public class BpmSyncHttpRequestTrigger extends BpmAbstractHttpRequestTrigger {
     @Resource
     private BpmProcessInstanceService processInstanceService;
-    @Resource
-    private RestTemplate restTemplate;
-
     @Override
     public BpmTriggerTypeEnum getType() {
         return BpmTriggerTypeEnum.HTTP_REQUEST;
@@ -56,40 +45,27 @@ public class BpmHttpRequestTrigger implements BpmTrigger {
             log.error("[execute][流程({}) HTTP 触发器请求配置为空]", processInstanceId);
             return;
         }
+
         // 2.1 设置请求头
         ProcessInstance processInstance = processInstanceService.getProcessInstance(processInstanceId);
-        Map<String, Object> processVariables = processInstance.getProcessVariables();
-        MultiValueMap<String, String> headers = new LinkedMultiValueMap<>();
-        headers.add(HEADER_TENANT_ID, processInstance.getTenantId());
-        SimpleModelUtils.addHttpRequestParam(headers, setting.getHeader(), processVariables);
+        MultiValueMap<String, String> headers = buildHttpHeaders(processInstance, setting.getHeader());
         // 2.2 设置请求体
-        MultiValueMap<String, String> body = new LinkedMultiValueMap<>();
-        SimpleModelUtils.addHttpRequestParam(body, setting.getBody(), processVariables);
-        body.add("processInstanceId", processInstanceId);
+        MultiValueMap<String, String> body = buildHttpBody(processInstance, setting.getBody());
-        // TODO @芋艿：要不要抽象一个 Http 请求的工具类，方便复用呢？
         // 3. 发起请求
-        HttpEntity<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(body, headers);
-        ResponseEntity<String> responseEntity;
-        try {
-            responseEntity = restTemplate.exchange(setting.getUrl(), HttpMethod.POST,
-                    requestEntity, String.class);
-            log.info("[execute][HTTP 触发器，请求头：{}，请求体：{}，响应结果：{}]", headers, body, responseEntity);
-        } catch (RestClientException e) {
-            log.error("[execute][HTTP 触发器，请求头：{}，请求体：{}，请求出错：{}]", headers, body, e.getMessage());
-            return;
-        }
+        ResponseEntity<String> responseEntity = sendHttpRequest(setting.getUrl(), headers, body);
         // 4.1 判断是否需要解析返回值
-        if (StrUtil.isEmpty(responseEntity.getBody())
+        if (responseEntity == null || StrUtil.isEmpty(responseEntity.getBody())
                 || !responseEntity.getStatusCode().is2xxSuccessful()
                 || CollUtil.isEmpty(setting.getResponse())) {
             return;
         }
         // 4.2 解析返回值, 返回值必须符合 CommonResult 规范。
         CommonResult<Map<String, Object>> respResult = JsonUtils.parseObjectQuietly(
-                responseEntity.getBody(), new TypeReference<>() {});
-        if (respResult == null || !respResult.isSuccess()){
+                responseEntity.getBody(), new TypeReference<>() {
+                });
+        if (respResult == null || !respResult.isSuccess()) {
             return;
         }
         // 4.3 获取需要更新的流程变量
@@ -103,11 +79,11 @@ public class BpmHttpRequestTrigger implements BpmTrigger {
     /**
      * 从请求返回值获取需要更新的流程变量
      *
-     * @param result 请求返回结果
+     * @param result           请求返回结果
      * @param responseSettings 返回设置
      * @return 需要更新的流程变量
      */
-    private Map<String, Object> getNeedUpdatedVariablesFromResponse(Map<String,Object> result,
+    private Map<String, Object> getNeedUpdatedVariablesFromResponse(Map<String, Object> result,
                                                                     List<KeyValue<String, String>> responseSettings) {
         Map<String, Object> updateVariables = new HashMap<>();
         if (CollUtil.isEmpty(result)) {