Skip to content

B
byit-myth-job
Commit 5bc94133 by huangfusuper
Options

参数包装

parent d38fa189
Showing with 99 additions and 10 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/TaskRunTheLifeCycleCallback.java
......@@ -19,7 +19,7 @@ public interface TaskRunTheLifeCycleCallback {
}
/**
* 后置处理器
* 后置处理器 postProcessBeforeInitialization
*
* @param jobTaskSchedule 任务对象
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/task/java/JavaParamWrappedCallback.java 0 → 100644
package com.byit.strategy.task.java;
import com.alibaba.fastjson.JSON;
import com.byit.annotations.MythRankOrder;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.enums.NodeTypeEnum;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.service.RunRecordingService;
import com.byit.strategy.TaskRunTheLifeCycleCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@Component
@Slf4j
@MythRankOrder(1)
public class JavaParamWrappedCallback implements TaskRunTheLifeCycleCallback {
private final RunRecordingService runRecordingService;
public JavaParamWrappedCallback(RunRecordingService runRecordingService) {
this.runRecordingService = runRecordingService;
}
/**
* 判断是否匹配类型
*
* @param jobTaskSchedule 任务对象
* @return 是否匹配本次的执行对象
*/
@Override
public boolean matchType(JobTaskSchedule jobTaskSchedule) {
return NodeTypeEnum.JAVA.getType().equals(jobTaskSchedule.getJobType());
}
/**
* 实例执行前
*
* @param jobTaskSchedule 任务对象
*/
@Override
public void postProcessAfterInitialization(JobTaskSchedule jobTaskSchedule) {
String runId = jobTaskSchedule.getRunId();
String flowName = jobTaskSchedule.getFlowName();
RunRecording runRecording = runRecordingService.findAllByRunIDAndFlowName(runId, flowName);
String extendedConfiguration = runRecording.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
String publicParam = flowExtendedConfiguration.getPublicParam();
Map<String,String> publicMap = JSON.parseObject(publicParam, Map.class);
if(publicMap == null) {
publicMap = new HashMap<>(8);
}
String runParam = jobTaskSchedule.getRunParam();
RunParamWrapped runParamWrapped = new RunParamWrapped();
runParamWrapped.setPrivateParam(runParam);
runParamWrapped.setPublicParamMap(publicMap);
jobTaskSchedule.setRunParam(JSON.toJSONString(runParamWrapped));
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/task/MakeUpScriptTaskCallbackRunTheLifeCycleCallback.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/task/script/ScriptMakeUpScriptTaskCallbackRunTheLifeCycleCallback.java
package com.byit.strategy.task;
package com.byit.strategy.task.script;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
......@@ -25,11 +25,11 @@ import java.util.Map;
@Component
@Slf4j
@MythRankOrder(2)
public class MakeUpScriptTaskCallbackRunTheLifeCycleCallback implements TaskRunTheLifeCycleCallback {
public class ScriptMakeUpScriptTaskCallbackRunTheLifeCycleCallback implements TaskRunTheLifeCycleCallback {
public static final String SCRIPT = "SCRIPT";
private final RunRecordingMapper runRecordingService;
public MakeUpScriptTaskCallbackRunTheLifeCycleCallback(RunRecordingMapper runRecordingService) {
public ScriptMakeUpScriptTaskCallbackRunTheLifeCycleCallback(RunRecordingMapper runRecordingService) {
this.runRecordingService = runRecordingService;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/task/ParamWrappedCallback.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/task/script/ScriptParamWrappedCallback.java
package com.byit.strategy.task;
package com.byit.strategy.task.script;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.annotations.MythRankOrder;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.enums.NodeTypeEnum;
import com.byit.model.JobTaskSchedule;
import com.byit.strategy.TaskRunTheLifeCycleCallback;
import lombok.extern.slf4j.Slf4j;
......@@ -22,7 +23,9 @@ import java.util.Map;
@Component
@Slf4j
@MythRankOrder(1)
public class ParamWrappedCallback implements TaskRunTheLifeCycleCallback {
public class ScriptParamWrappedCallback implements TaskRunTheLifeCycleCallback {
public static final String SCRIPT = "SCRIPT";
/**
* 实例执行前
......@@ -64,6 +67,12 @@ public class ParamWrappedCallback implements TaskRunTheLifeCycleCallback {
*/
@Override
public boolean matchType(JobTaskSchedule jobTaskSchedule) {
return true;
String jobType = jobTaskSchedule.getJobType();
NodeTypeEnum typeByCode = NodeTypeEnum.getTypeByCode(jobType);
if (typeByCode == null) {
throw new RuntimeException(String.format("调度暂不支持此种类型的任务:%s", jobType));
}
//是否是脚本类型的
return typeByCode.getType().equals(SCRIPT);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaNodeExecutorTask.java
......@@ -2,6 +2,7 @@ package com.byit.task;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.BeanStrategyPackage;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.dto.plugin.RunLog;
......@@ -21,6 +22,8 @@ import com.byit.service.FlowStatusService;
import com.byit.service.RunRecordingService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunJavaServiceImpl;
import com.byit.strategy.TaskRunTheLifeCycleCallback;
import com.byit.util.ClassSortUtil;
import com.byit.util.GetRegConfig;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
......@@ -80,6 +83,17 @@ public class JavaNodeExecutorTask implements TimerTask {
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLogWithBLOBs jobTaskRunLogById = null;
try {
//获取该节点的全部声明周期函数
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(mythJobTaskSchedule)) {
beanStrategyPackageBean.postProcessAfterInitialization(mythJobTaskSchedule);
}
});
String runParam = mythJobTaskSchedule.getRunParam();
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
if (runParamWrapped == null) {
......
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