设置结束回调方法

byit-myth-core/myth-admin-core/src/main/java/com/byit/dto/RunRecordingWrapped.java

@@ -2,7 +2,9 @@ package com.byit.dto;
 
 import com.byit.model.JobTask;
 import com.byit.model.RunRecording;
+import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 import java.io.Serializable;
 import java.util.List;
@@ -14,6 +16,8 @@ import java.util.List;
  * @date 2020年11月25日18:31:07
  */
 @Data
+@AllArgsConstructor
+@NoArgsConstructor
 public class RunRecordingWrapped implements Serializable {
     private static final long serialVersionUID = 6309696511670011734L;
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/impl/RpcCallbackInstanceRunTheLifeCycleCallback.java

@@ -18,7 +18,9 @@ import com.byit.param.CommunicationParam;
 import com.byit.service.WorkspaceService;
 import com.byit.strategy.InstanceRunTheLifeCycleCallback;
 import com.byit.task.annotations.TaskClient;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -31,15 +33,19 @@ import java.util.stream.Collectors;
  * @author huangfu
  */
 @Component
+@Slf4j
 public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTheLifeCycleCallback {
 
+    public static final String EXTENDED_PARAMS_S = "extended:params:%s";
     @TaskClient(timeout = 30000, callMethodEnum = CallMethodEnum.SYNCHRONIZE)
     private TaskServer taskServer;
 
     private final WorkspaceService workspaceService;
+    private final StringRedisTemplate stringRedisTemplate;
 
-    public RpcCallbackInstanceRunTheLifeCycleCallback(WorkspaceService workspaceService) {
+    public RpcCallbackInstanceRunTheLifeCycleCallback(WorkspaceService workspaceService, StringRedisTemplate stringRedisTemplate) {
         this.workspaceService = workspaceService;
+        this.stringRedisTemplate = stringRedisTemplate;
     }
 
     /**
@@ -51,10 +57,22 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
      */
     @Override
     public RunRecordingWrapped postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList) {
-        PluginRpcRequestPacket param = buildPluginRpcRequestPacket(runRecording);
+        String extendedConfiguration = runRecording.getExtendedConfiguration();
+        FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
+        if(flowExtendedConfiguration == null){
+            return new RunRecordingWrapped(jobTaskList,runRecording);
+        }
+        String rpcServerKey = flowExtendedConfiguration.getRpcStartServerKey();
+        if(StringUtils.isBlank(rpcServerKey)){
+            return new RunRecordingWrapped(jobTaskList,runRecording);
+        }
+        CallbackDto callbackDto = new CallbackDto();
+        PluginRpcRequestPacket param = buildPluginRpcRequestPacket(runRecording,callbackDto,rpcServerKey);
         PluginRpcResponsePacket call = taskServer.call(param);
         Object result = call.getResult();
         ScriptParamAndPlaceholderDto rpcScriptParamAndPlaceholderDto = (ScriptParamAndPlaceholderDto) result;
+        //放置到对应的redis里面
+        stringRedisTemplate.opsForValue().set(String.format(EXTENDED_PARAMS_S,runRecording.getRunId()),JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
         if (rpcScriptParamAndPlaceholderDto != null) {
             List<JobTask> jobTasks = jobTaskList.stream().map(jobTask -> {
                 //新增参数
@@ -131,19 +149,14 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
      * @param runRecording 执行实例
      * @return {@link PluginRpcRequestPacket}
      */
-    private PluginRpcRequestPacket buildPluginRpcRequestPacket(RunRecording runRecording) {
+    private PluginRpcRequestPacket buildPluginRpcRequestPacket(RunRecording runRecording, CallbackDto callbackDto,String serverKey) {
         PluginRpcRequestPacket rpcRequestPacket = new PluginRpcRequestPacket();
-        String extendedConfiguration = runRecording.getExtendedConfiguration();
-        FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
-        String rpcServerKey = flowExtendedConfiguration.getRpcServerKey();
         rpcRequestPacket.setCallMethodEnum(CallMethodEnum.SYNCHRONIZE);
-        rpcRequestPacket.setJobName(rpcServerKey);
+        rpcRequestPacket.setJobName(serverKey);
         CommunicationParam param = new CommunicationParam();
         param.setRunId(runRecording.getRunId());
         param.setHasMakeUp(String.valueOf(runRecording.getScheduleType()));
 
-
-        CallbackDto callbackDto = new CallbackDto();
         callbackDto.setFlowName(runRecording.getFlowName());
         callbackDto.setVersionName(runRecording.getFlowVersionName());
 
@@ -163,5 +176,21 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
     @Override
     public void postProcessBeforeInitialization(RunRecording runRecording) {
 
+        String extendedConfiguration = runRecording.getExtendedConfiguration();
+        FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
+        if(flowExtendedConfiguration == null){
+            return;
+        }
+        String rpcEndServerKey = flowExtendedConfiguration.getRpcEndServerKey();
+        if(StringUtils.isBlank(rpcEndServerKey)){
+            return;
+        }
+        String extendedParam = stringRedisTemplate.opsForValue().get(String.format(EXTENDED_PARAMS_S, runRecording.getRunId()));
+
+        CallbackDto callbackDto = new CallbackDto();
+        callbackDto.setData(extendedParam);
+        PluginRpcRequestPacket rpcRequestPacket = buildPluginRpcRequestPacket(runRecording, callbackDto, rpcEndServerKey);
+        PluginRpcResponsePacket call = taskServer.call(rpcRequestPacket);
+        log.info("回调{}通讯成功！",call);
     }
 }

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/common/CallbackDto.java

@@ -17,4 +17,5 @@ public class CallbackDto implements Serializable {
     private String flowName;
     private String workspaceName;
     private String versionName;
+    private String data;
 }

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/FlowExtendedConfiguration.java

@@ -10,5 +10,6 @@ import lombok.Data;
  */
 @Data
 public class FlowExtendedConfiguration {
-    private String rpcServerKey;
+    private String rpcStartServerKey;
+    private String rpcEndServerKey;
 }