Merge remote-tracking branch 'origin/develop-many-instance' into develop-many-instance

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowServiceImpl.java

@@ -431,6 +431,42 @@ public class FlowServiceImpl implements FlowService {
         ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
         JSONObject jsonObject = JSON.parseObject(param);
         String runId = jsonObject.getString("runId");
+        //查找到运行批次
+        List<RunRecording> runRecordingList = runRecordingMapper.findByRunID(runId);
+        //查看是否存在运行中的实例
+        List<RunRecording> collect = runRecordingList.stream().filter(runRecording -> {
+            String flowStatus = runRecording.getFlowStatus();
+            return ScheduleStatusEnum.STOP.getCode().equals(flowStatus) || ScheduleStatusEnum.STARTING.getCode().equals(flowStatus);
+        }).collect(Collectors.toList());
+        ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(collect), "没有正在运行中的实例，请刷新页面后重试！");
+        //遍历运行批次  将实例更改为快速失败
+        //释放该工作流下所由暂停的节点
+        collect.forEach(runRecording -> {
+            runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
+            runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
+            if (ScheduleStatusEnum.STOP.getCode().equals(runRecording.getFlowStatus())) {
+
+                runRecording.setFlowStatus(ScheduleStatusEnum.STARTING.getCode());
+            }
+
+            List<JobTask> byRunId = jobTaskService.findByRunId(runId);
+            List<JobTask> stopTask = byRunId.stream().filter(task -> JobTriggerStatusEnums.STOP.getCode().equals(task.getTriggerStatus())).collect(Collectors.toList());
+            stopTask.forEach(task -> {
+                task.setTriggerStatus(JobTriggerStatusEnums.START.getCode());
+                jobTaskService.update(task);
+            });
+            runRecordingMapper.updateRunRecordingById(runRecording);
+
+        });
+        return  new KillDto(true, "操作成功！");
+    }
+
+
+    @Deprecated
+    public KillDto killFlow111(String param) throws InterruptedException {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonObject = JSON.parseObject(param);
+        String runId = jsonObject.getString("runId");
         ValidationUtil.dataNotBank(runId, "运行实例id不允许为空！");
         String workspaceName = jsonObject.getString("workspaceName");
         ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/TaskAndLogServerImpl.java

@@ -85,13 +85,16 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
         Map<String, TaskRunTheLifeCycleCallback> stringTaskRunTheLifeCycleCallbackMap = SpringUtil.getBeansOfType(TaskRunTheLifeCycleCallback.class);
         //数据排序
         List<BeanStrategyPackage<TaskRunTheLifeCycleCallback>> beanStrategyPackages = ClassSortUtil.objectSort(stringTaskRunTheLifeCycleCallbackMap);
-        beanStrategyPackages.forEach(beanStrategyPackage -> {
-            log.info("--------------脚本节点生命周期开始回调{}--------------", beanStrategyPackage.getBeanName());
-            TaskRunTheLifeCycleCallback beanStrategyPackageBean = beanStrategyPackage.getBean();
-            if (beanStrategyPackageBean.matchType(jobTaskSchedule)) {
-                beanStrategyPackageBean.postProcessAfterInitialization(jobTaskSchedule);
-            }
-        });
+        if(!FlowPropertyEnum.IS_INNER.getCode().equals(jobTask.getIsVirtual())) {
+            beanStrategyPackages.forEach(beanStrategyPackage -> {
+                log.info("--------------脚本节点生命周期开始回调{}--------------", beanStrategyPackage.getBeanName());
+                TaskRunTheLifeCycleCallback beanStrategyPackageBean = beanStrategyPackage.getBean();
+                if (beanStrategyPackageBean.matchType(jobTaskSchedule)) {
+                    beanStrategyPackageBean.postProcessAfterInitialization(jobTaskSchedule);
+                }
+            });
+        }
+
 
         BeanUtils.copyProperties(jobTaskSchedule, jobTaskRunLog);
         jobTaskRunLog.setRunId(jobTaskSchedule.getRunId());
@@ -154,6 +157,7 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
                     .isInner(FlowPropertyEnum.IS_INNER.getCode())
                     .failFast(RunRecordingEnum.FAIL_FAST_YES.getCode())
                     .workspaceId(virFlow.getWorkspaceId())
+                    .extendedConfiguration(virFlow.getExtendedConfiguration())
                     .build();
             runRecordingService.saveRunRecording(runRecording);
             log.error("在将该虚节点对应的节点拉取过来");

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobTaskThreadRunHelper.java

@@ -106,8 +106,13 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                         //删除这个数据 并且添加到日志
                         try {
                             taskAndLogServer.addRunLogAndRemoveTask(jobTask,true,false);
-                        } catch (UnknownHostException e) {
+                        } catch (Exception e) {
                             e.printStackTrace();
+                            try {
+                                taskAndLogServer.addRunLogAndRemoveTask(jobTask,true,true, MythLogUtils.getMessage(e));
+                            } catch (UnknownHostException unknownHostException) {
+                                unknownHostException.printStackTrace();
+                            }
                         }
                     } catch (Exception e) {
                         log.error("-----------------节点执行时出现异常,{}----------", MythLogUtils.getMessage(e));
@@ -137,12 +142,17 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                         //删除这个数据 并且添加到日志
                         try {
                             taskAndLogServer.addRunLogAndRemoveTask(jobTask,false,false);
-                        } catch (UnknownHostException e) {
+                        } catch (Exception e) {
                             log.error("-----------------节点执行时出现异常,{}----------",MythLogUtils.getMessage(e));
+                            try {
+                                taskAndLogServer.addRunLogAndRemoveTask(jobTask,false,true, MythLogUtils.getMessage(e));
+                            } catch (UnknownHostException unknownHostException) {
+                                unknownHostException.printStackTrace();
+                            }
                         }
                     }catch (Exception businessException) {
                         try {
-                            taskAndLogServer.addRunLogAndRemoveTask(jobTask,false,true);
+                            taskAndLogServer.addRunLogAndRemoveTask(jobTask,false,true,MythLogUtils.getMessage(businessException));
                         } catch (UnknownHostException e) {
                             log.error("-----------------节点执行时出现异常,{}----------",MythLogUtils.getMessage(e));
                         }