Commit ac25edc7 by huangfusuper

实例声明周期回调失败,那么实例失败

parent 993c2431
......@@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSON;
import com.byit.dto.NodeRelyDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.dto.specials.MakeUpReturn;
import com.byit.dto.specials.RepairFlow;
import com.byit.dto.specials.RepairTimeParam;
......@@ -209,7 +210,7 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
}
if (CollectionUtil.isNotEmpty(nodeRelyDtoSet)) {
//构建等待队列
List<WaitingRecord> waitingRecords = buildWaitingRecordList(flowByName, nodeRelyDtoSet.size(), operator, repairTimeList);
List<WaitingRecord> waitingRecords = buildWaitingRecordList(flowByName, nodeRelyDtoSet.size(), operator, repairTimeList,JSON.toJSONString(publicParam));
waitingRecords.forEach(waitingRecordMapper::insertSelective);
List<RunRecording> runRecordings = buildRunRecordingList(waitingRecords);
List<WaitingTask> waitingTaskList = buildWaitingTaskResult(waitingRecords, nodeRelyDtoSet, specialJobParam.getPublicParam(), timeFormatName, nowTimeFormatName);
......@@ -354,7 +355,7 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
* @param repairTimeList 补时间
* @return 等待队列集合
*/
private List<WaitingRecord> buildWaitingRecordList(Flow flow, Integer nodeCount, String operator, List<RepairTimeParam> repairTimeList) {
private List<WaitingRecord> buildWaitingRecordList(Flow flow, Integer nodeCount, String operator, List<RepairTimeParam> repairTimeList, String publicParam) {
//查看当前排队的工作流最大排队序号
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null) {
......@@ -368,6 +369,15 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
BeanUtils.copyProperties(flow, waitingRecord);
//设置排期
BeanUtils.copyProperties(flow, waitingRecord);
String extendedConfiguration = flow.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
if(flowExtendedConfiguration != null){
flowExtendedConfiguration.setPublicParam(publicParam);
}else{
flowExtendedConfiguration = new FlowExtendedConfiguration();
flowExtendedConfiguration.setPublicParam(publicParam);
}
waitingRecord.setExtendedConfiguration(JSON.toJSONString(flowExtendedConfiguration));
waitingRecord.setFlowVersionName(flow.getVersionName());
waitingRecord.setRunId(IDGenerationStrategy.runIdGenerationStrategy(serverPort));
waitingRecord.setFlowNodeCount(nodeCount);
......
......@@ -2,7 +2,9 @@ package com.byit.service.impl;
import com.byit.dto.RunRecordingWrapped;
import com.byit.enums.NodeNameEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.event.FlowScanEndEvent;
import com.byit.job.utils.CronExpression;
import com.byit.model.Flow;
......@@ -108,11 +110,17 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
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();
try {
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(build, jobTasks);
build = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
}
}catch (Exception e) {
e.printStackTrace();
build.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
build.setFlowRunResult(RunResultEnum.RUN_ERROR.getCode());
}
//开始正式保存
runRecordingService.saveRunRecording(build);
......
......@@ -3,6 +3,7 @@ package com.byit.service.mapservice.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.dto.RunRecordingWrapped;
import com.byit.enums.*;
import com.byit.enums.task.RunResultEnum;
import com.byit.job.utils.CronExpression;
import com.byit.job.utils.DateUtil;
import com.byit.model.*;
......@@ -269,12 +270,18 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
//开始执行生命周期 开始的
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();
try {
//回调生命周期
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(runRecording, jobTasks);
runRecording = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
}
}catch (Exception e) {
e.printStackTrace();
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecording.setFlowRunResult(RunResultEnum.RUN_ERROR.getCode());
}
runRecordingService.updateRunRecordingById(runRecording);
......
......@@ -24,6 +24,7 @@ import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
......@@ -188,9 +189,31 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
return;
}
String extendedParam = stringRedisTemplate.opsForValue().get(String.format(EXTENDED_PARAMS_S, runRecording.getRunId()));
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(extendedParam, ScriptParamAndPlaceholderDto.class);
String publicParam = flowExtendedConfiguration.getPublicParam();
Map<String,String> callData = new HashMap<>(2);
if(StringUtils.isNoneBlank(publicParam)){
Map<String, String> stringStringMap = (Map<String, String>) JSON.parse(publicParam);
if(CollectionUtil.isNotEmpty(stringStringMap)) {
callData.putAll(stringStringMap);
}
}
if(scriptParamAndPlaceholderDto != null){
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
Map<String, String> placeholder = scriptParamAndPlaceholderDto.getPlaceholder();
if(CollectionUtil.isNotEmpty(param)) {
callData.putAll(param);
}
if(CollectionUtil.isNotEmpty(placeholder)) {
callData.putAll(placeholder);
}
}
CallbackDto callbackDto = new CallbackDto();
callbackDto.setData(extendedParam);
callbackDto.setData(JSON.toJSONString(callData));
PluginRpcRequestPacket rpcRequestPacket = buildPluginRpcRequestPacket(runRecording, callbackDto, rpcEndServerKey);
try {
PluginRpcResponsePacket call = taskServer.call(rpcRequestPacket);
......
......@@ -10,6 +10,17 @@ import lombok.Data;
*/
@Data
public class FlowExtendedConfiguration {
/**
* rpc开始节点的回调
*/
private String rpcStartServerKey;
/**
* rpc结束节点的回调
*/
private String rpcEndServerKey;
/**
* 公共参数
*/
public String publicParam;
}
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