Commit 40aeadac by huangfusuper

增加生命周期回调 生成实例之前的回调

parent ed07e73f
package com.byit.service.impl; package com.byit.service.impl;
import com.byit.dto.RunRecordingWrapped;
import com.byit.enums.NodeNameEnum; import com.byit.enums.NodeNameEnum;
import com.byit.enums.ScheduleTypeEnum; import com.byit.enums.ScheduleTypeEnum;
import com.byit.event.FlowScanEndEvent; import com.byit.event.FlowScanEndEvent;
...@@ -9,7 +10,9 @@ import com.byit.model.JobTask; ...@@ -9,7 +10,9 @@ import com.byit.model.JobTask;
import com.byit.model.Node; import com.byit.model.Node;
import com.byit.model.RunRecording; import com.byit.model.RunRecording;
import com.byit.service.*; import com.byit.service.*;
import com.byit.strategy.InstanceRunTheLifeCycleCallback;
import com.byit.util.IDGenerationStrategy; import com.byit.util.IDGenerationStrategy;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
...@@ -21,10 +24,7 @@ import org.springframework.transaction.annotation.Propagation; ...@@ -21,10 +24,7 @@ import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.text.ParseException; import java.text.ParseException;
import java.util.ArrayList; import java.util.*;
import java.util.Date;
import java.util.List;
import java.util.UUID;
/** /**
* @author huangfu * @author huangfu
...@@ -69,14 +69,14 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis ...@@ -69,14 +69,14 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
build.setDispatchIp("0.0.0.0"); build.setDispatchIp("0.0.0.0");
build.setFlowVersionName(flow.getVersionName()); build.setFlowVersionName(flow.getVersionName());
build.setTriggerTime(flow.getTriggerNextTime()); build.setTriggerTime(flow.getTriggerNextTime());
runRecordingService.saveRunRecording(build);
boolean isScheduleFollow = "1".equals(flow.getScheduleFollow()); boolean isScheduleFollow = "1".equals(flow.getScheduleFollow());
log.info("-------------【开始保存节点信息,任务是否为跟随工作流,{}】---------------",isScheduleFollow); log.info("-------------【开始保存节点信息,任务是否为跟随工作流,{}】---------------",isScheduleFollow);
List<JobTask> jobTasks = new ArrayList<>(32); List<JobTask> jobTasks = new ArrayList<>(32);
nodes.forEach(node -> { for (Node node : nodes) {
JobTask jobTask = new JobTask(); JobTask jobTask = new JobTask();
jobTask.setScheduleType(1); jobTask.setScheduleType(1);
jobTask.setTriggerStatus("1"); jobTask.setTriggerStatus("1");
...@@ -102,8 +102,23 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis ...@@ -102,8 +102,23 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
jobTask.setNodeDepend(parentIds); jobTask.setNodeDepend(parentIds);
} }
jobTasks.add(jobTask); jobTasks.add(jobTask);
}); }
Map<String, InstanceRunTheLifeCycleCallback> beansOfType = SpringUtil.getBeansOfType(InstanceRunTheLifeCycleCallback.class);
Set<Map.Entry<String, InstanceRunTheLifeCycleCallback>> entries = beansOfType.entrySet();
//回调生命周期
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(build, jobTasks);
build = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
}
//开始正式保存
runRecordingService.saveRunRecording(build);
jobTaskService.saveJobTasks(jobTasks); jobTaskService.saveJobTasks(jobTasks);
log.info("-------------【开始修改工作流{}的下次运行时间,以及各种状态】---------------",flow); log.info("-------------【开始修改工作流{}的下次运行时间,以及各种状态】---------------",flow);
try { try {
Long triggerNextTime = flow.getTriggerNextTime(); Long triggerNextTime = flow.getTriggerNextTime();
......
package com.byit.service.mapservice.impl; package com.byit.service.mapservice.impl;
import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.collection.CollectionUtil;
import com.byit.dto.RunRecordingWrapped;
import com.byit.enums.*; import com.byit.enums.*;
import com.byit.job.utils.CronExpression; import com.byit.job.utils.CronExpression;
import com.byit.job.utils.DateUtil; import com.byit.job.utils.DateUtil;
import com.byit.model.*; import com.byit.model.*;
import com.byit.service.*; import com.byit.service.*;
import com.byit.service.mapservice.RunRecordingAndJobTaskService; import com.byit.service.mapservice.RunRecordingAndJobTaskService;
import com.byit.strategy.InstanceRunTheLifeCycleCallback;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
...@@ -17,9 +20,7 @@ import org.springframework.transaction.annotation.Transactional; ...@@ -17,9 +20,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.text.ParseException; import java.text.ParseException;
import java.util.ArrayList; import java.util.*;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
...@@ -245,7 +246,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask ...@@ -245,7 +246,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
runRecording.setStartTime(new Date()); runRecording.setStartTime(new Date());
//将运行实例改为以执行 //将运行实例改为以执行
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode()); runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode());
runRecordingService.updateRunRecordingById(runRecording);
log.info("-------查询等待节点成功,查询对应的等待节点成功,开始保存对应的等待节点{}-------",allByWaitId); log.info("-------查询等待节点成功,查询对应的等待节点成功,开始保存对应的等待节点{}-------",allByWaitId);
List<Integer> ids = new ArrayList<>(2); List<Integer> ids = new ArrayList<>(2);
...@@ -263,11 +264,24 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask ...@@ -263,11 +264,24 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
target.setTriggerTime(0L); target.setTriggerTime(0L);
return target; return target;
}).collect(Collectors.toList()); }).collect(Collectors.toList());
//开始执行生命周期 开始的
Map<String, InstanceRunTheLifeCycleCallback> beansOfType = SpringUtil.getBeansOfType(InstanceRunTheLifeCycleCallback.class);
Set<Map.Entry<String, InstanceRunTheLifeCycleCallback>> entries = beansOfType.entrySet();
//回调生命周期
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(runRecording, jobTasks);
runRecording = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
}
runRecordingService.updateRunRecordingById(runRecording);
jobTaskService.saveJobTasks(jobTasks); jobTaskService.saveJobTasks(jobTasks);
//保存等待实例 //保存等待实例
waitingRecord.setWaitOrder(-1); waitingRecord.setWaitOrder(-1);
log.info("-------保存jobTask成功,开始修改等待实例{}-----------------",waitingRecord); log.info("-------保存jobTask成功,开始修改等待实例{}-----------------",waitingRecord);
//waitingRecordService.updateById(waitingRecord);
if (CollectionUtil.isNotEmpty(ids)) { if (CollectionUtil.isNotEmpty(ids)) {
taskService.deleteAllByIds(ids); taskService.deleteAllByIds(ids);
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment