| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.dolphinscheduler.api.service.impl; |
| |
| import static java.util.stream.Collectors.toSet; |
| import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_CONDITIONS; |
| import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_DEPENDENT; |
| import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SUB_PROCESS; |
| |
| import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationCreateRequest; |
| import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationFilterRequest; |
| import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationUpdateUpstreamRequest; |
| import org.apache.dolphinscheduler.api.enums.Status; |
| import org.apache.dolphinscheduler.api.exceptions.ServiceException; |
| import org.apache.dolphinscheduler.api.service.ProcessTaskRelationService; |
| import org.apache.dolphinscheduler.api.service.ProjectService; |
| import org.apache.dolphinscheduler.common.constants.Constants; |
| import org.apache.dolphinscheduler.common.enums.ConditionType; |
| import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
| import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionLog; |
| import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation; |
| import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog; |
| import org.apache.dolphinscheduler.dao.entity.Project; |
| import org.apache.dolphinscheduler.dao.entity.TaskDefinition; |
| import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog; |
| import org.apache.dolphinscheduler.dao.entity.User; |
| import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionLogMapper; |
| import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper; |
| import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationLogMapper; |
| import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper; |
| import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; |
| import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper; |
| import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper; |
| import org.apache.dolphinscheduler.service.process.ProcessService; |
| |
| import org.apache.commons.collections4.CollectionUtils; |
| import org.apache.commons.lang3.StringUtils; |
| |
| import java.util.ArrayList; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| |
| import lombok.extern.slf4j.Slf4j; |
| |
| import org.springframework.beans.factory.annotation.Autowired; |
| import org.springframework.stereotype.Service; |
| import org.springframework.transaction.annotation.Transactional; |
| |
| import com.baomidou.mybatisplus.core.metadata.IPage; |
| import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
| import com.google.common.collect.Lists; |
| |
| /** |
| * process task relation service impl |
| */ |
| @Service |
| @Slf4j |
| public class ProcessTaskRelationServiceImpl extends BaseServiceImpl implements ProcessTaskRelationService { |
| |
| @Autowired |
| private ProjectMapper projectMapper; |
| |
| @Autowired |
| private ProjectService projectService; |
| |
| @Autowired |
| private ProcessTaskRelationMapper processTaskRelationMapper; |
| |
| @Autowired |
| private TaskDefinitionLogMapper taskDefinitionLogMapper; |
| |
| @Autowired |
| private TaskDefinitionMapper taskDefinitionMapper; |
| |
| @Autowired |
| private ProcessDefinitionMapper processDefinitionMapper; |
| |
| @Autowired |
| private ProcessDefinitionLogMapper processDefinitionLogMapper; |
| |
| @Autowired |
| private ProcessService processService; |
| |
| @Autowired |
| private ProcessTaskRelationLogMapper processTaskRelationLogMapper; |
| |
| /** |
| * create process task relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param processDefinitionCode processDefinitionCode |
| * @param preTaskCode preTaskCode |
| * @param postTaskCode postTaskCode |
| * @return create result code |
| */ |
| @Transactional |
| @Override |
| public Map<String, Object> createProcessTaskRelation(User loginUser, long projectCode, long processDefinitionCode, |
| long preTaskCode, long postTaskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode); |
| if (processDefinition == null) { |
| log.error("Process definition does not exist, processCode:{}.", processDefinitionCode); |
| putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode)); |
| return result; |
| } |
| if (processDefinition.getProjectCode() != projectCode) { |
| log.error("Process definition's project does not match project {}.", projectCode); |
| putMsg(result, Status.PROJECT_PROCESS_NOT_MATCH); |
| return result; |
| } |
| updateProcessDefiniteVersion(loginUser, result, processDefinition); |
| List<ProcessTaskRelation> processTaskRelationList = |
| processTaskRelationMapper.queryByProcessCode(processDefinitionCode); |
| List<ProcessTaskRelation> processTaskRelations = Lists.newArrayList(processTaskRelationList); |
| if (!processTaskRelations.isEmpty()) { |
| Map<Long, ProcessTaskRelation> preTaskCodeMap = |
| processTaskRelations.stream().filter(r -> r.getPostTaskCode() == postTaskCode) |
| .collect(Collectors.toMap(ProcessTaskRelation::getPreTaskCode, |
| processTaskRelation -> processTaskRelation)); |
| if (!preTaskCodeMap.isEmpty()) { |
| if (preTaskCodeMap.containsKey(preTaskCode) || (!preTaskCodeMap.containsKey(0L) && preTaskCode == 0L)) { |
| putMsg(result, Status.PROCESS_TASK_RELATION_EXIST, String.valueOf(processDefinitionCode)); |
| return result; |
| } |
| if (preTaskCodeMap.containsKey(0L) && preTaskCode != 0L) { |
| // delete no upstream |
| processTaskRelations.remove(preTaskCodeMap.get(0L)); |
| } |
| } |
| } |
| TaskDefinition postTaskDefinition = taskDefinitionMapper.queryByCode(postTaskCode); |
| ProcessTaskRelation processTaskRelation = setRelation(processDefinition, postTaskDefinition); |
| if (preTaskCode != 0L) { |
| TaskDefinition preTaskDefinition = taskDefinitionMapper.queryByCode(preTaskCode); |
| List<ProcessTaskRelation> upstreamTaskRelationList = processTaskRelations.stream() |
| .filter(r -> r.getPostTaskCode() == preTaskCode).collect(Collectors.toList()); |
| // upstream is or not exist |
| if (upstreamTaskRelationList.isEmpty()) { |
| ProcessTaskRelation preProcessTaskRelation = setRelation(processDefinition, preTaskDefinition); |
| preProcessTaskRelation.setPreTaskCode(0L); |
| preProcessTaskRelation.setPreTaskVersion(0); |
| processTaskRelations.add(preProcessTaskRelation); |
| } |
| processTaskRelation.setPreTaskCode(preTaskDefinition.getCode()); |
| processTaskRelation.setPreTaskVersion(preTaskDefinition.getVersion()); |
| } else { |
| processTaskRelation.setPreTaskCode(0L); |
| processTaskRelation.setPreTaskVersion(0); |
| } |
| processTaskRelations.add(processTaskRelation); |
| updateRelation(loginUser, result, processDefinition, processTaskRelations); |
| return result; |
| } |
| |
| private ProcessTaskRelationLog persist2ProcessTaskRelationLog(User user, ProcessTaskRelation processTaskRelation) { |
| ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(processTaskRelation); |
| processTaskRelationLog.setOperator(user.getId()); |
| processTaskRelationLog.setOperateTime(new Date()); |
| int result = processTaskRelationLogMapper.insert(processTaskRelationLog); |
| if (result <= 0) { |
| throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_LOG_ERROR, |
| processTaskRelationLog.getPreTaskCode(), processTaskRelationLog.getPostTaskCode()); |
| } |
| return processTaskRelationLog; |
| } |
| |
| private List<ProcessTaskRelationLog> batchPersist2ProcessTaskRelationLog(User user, |
| List<ProcessTaskRelation> processTaskRelations) { |
| Date now = new Date(); |
| List<ProcessTaskRelationLog> processTaskRelationLogs = new ArrayList<>(); |
| |
| for (ProcessTaskRelation processTaskRelation : processTaskRelations) { |
| ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(processTaskRelation); |
| processTaskRelationLog.setOperator(user.getId()); |
| processTaskRelationLog.setOperateTime(now); |
| processTaskRelationLogs.add(processTaskRelationLog); |
| } |
| int result = processTaskRelationLogMapper.batchInsert(processTaskRelationLogs); |
| if (result != processTaskRelationLogs.size()) { |
| throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_LOG_ERROR); |
| } |
| return processTaskRelationLogs; |
| } |
| |
| private void updateVersions(ProcessTaskRelation processTaskRelation) { |
| // workflow |
| ProcessDefinition processDefinition = |
| processDefinitionMapper.queryByCode(processTaskRelation.getProcessDefinitionCode()); |
| processTaskRelation.setProcessDefinitionVersion(processDefinition.getVersion()); |
| |
| // tasks |
| TaskDefinition preTaskDefinition = taskDefinitionMapper.queryByCode(processTaskRelation.getPreTaskCode()); |
| processTaskRelation.setPreTaskVersion(preTaskDefinition.getVersion()); |
| TaskDefinition postTaskDefinition = taskDefinitionMapper.queryByCode(processTaskRelation.getPostTaskCode()); |
| processTaskRelation.setPostTaskVersion(postTaskDefinition.getVersion()); |
| } |
| |
| /** |
| * create resource process task relation |
| * |
| * @param loginUser login user |
| * @param taskRelationCreateRequest project code |
| * @return ProcessTaskRelation object |
| */ |
| @Override |
| @Transactional |
| public ProcessTaskRelation createProcessTaskRelationV2(User loginUser, |
| TaskRelationCreateRequest taskRelationCreateRequest) { |
| ProcessTaskRelation processTaskRelation = taskRelationCreateRequest.convert2ProcessTaskRelation(); |
| ProcessDefinition processDefinition = |
| processDefinitionMapper.queryByCode(processTaskRelation.getProcessDefinitionCode()); |
| if (processDefinition == null) { |
| throw new ServiceException(Status.PROCESS_DEFINE_NOT_EXIST, |
| String.valueOf(processTaskRelation.getProcessDefinitionCode())); |
| } |
| if (processTaskRelation.getProjectCode() == 0) { |
| processTaskRelation.setProjectCode(processDefinition.getProjectCode()); |
| } |
| Project project = projectMapper.queryByCode(processTaskRelation.getProjectCode()); |
| projectService.checkProjectAndAuthThrowException(loginUser, project, null); |
| |
| // persistence process task relation and process task relation log to database |
| this.updateVersions(processTaskRelation); |
| int insert = processTaskRelationMapper.insert(processTaskRelation); |
| if (insert <= 0) { |
| throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR, processTaskRelation.getPreTaskCode(), |
| processTaskRelation.getPostTaskCode()); |
| } |
| this.persist2ProcessTaskRelationLog(loginUser, processTaskRelation); |
| |
| return processTaskRelation; |
| } |
| |
| private ProcessTaskRelation setRelation(ProcessDefinition processDefinition, TaskDefinition taskDefinition) { |
| Date now = new Date(); |
| ProcessTaskRelation processTaskRelation = new ProcessTaskRelation(); |
| processTaskRelation.setProjectCode(processDefinition.getProjectCode()); |
| processTaskRelation.setProcessDefinitionCode(processDefinition.getCode()); |
| processTaskRelation.setProcessDefinitionVersion(processDefinition.getVersion()); |
| processTaskRelation.setPostTaskCode(taskDefinition.getCode()); |
| processTaskRelation.setPostTaskVersion(taskDefinition.getVersion()); |
| processTaskRelation.setConditionType(ConditionType.NONE); |
| processTaskRelation.setConditionParams("{}"); |
| processTaskRelation.setCreateTime(now); |
| processTaskRelation.setUpdateTime(now); |
| return processTaskRelation; |
| } |
| |
| private void updateProcessDefiniteVersion(User loginUser, Map<String, Object> result, |
| ProcessDefinition processDefinition) { |
| int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, Boolean.TRUE, Boolean.TRUE); |
| if (insertVersion <= 0) { |
| log.error("Update process definition error, projectCode:{}, processDefinitionCode:{}.", |
| processDefinition.getProjectCode(), processDefinition.getCode()); |
| putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR); |
| throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR); |
| } else |
| log.info( |
| "Update process definition complete, new version is {}, projectCode:{}, processDefinitionCode:{}.", |
| insertVersion, processDefinition.getProjectCode(), processDefinition.getCode()); |
| processDefinition.setVersion(insertVersion); |
| } |
| |
| /** |
| * delete process task relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param processDefinitionCode process definition code |
| * @param taskCode the post task code |
| * @return delete result code |
| */ |
| @Transactional |
| @Override |
| public Map<String, Object> deleteTaskProcessRelation(User loginUser, long projectCode, long processDefinitionCode, |
| long taskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| if (taskCode == 0) { |
| log.error( |
| "Delete task process relation error due to parameter taskCode is 0, projectCode:{}, processDefinitionCode:{}.", |
| projectCode, processDefinitionCode); |
| putMsg(result, Status.DELETE_TASK_PROCESS_RELATION_ERROR); |
| return result; |
| } |
| ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode); |
| if (processDefinition == null) { |
| log.error("Process definition does not exist, processDefinitionCode:{}.", processDefinitionCode); |
| putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode)); |
| return result; |
| } |
| TaskDefinition taskDefinition = taskDefinitionMapper.queryByCode(taskCode); |
| if (null == taskDefinition) { |
| log.error("Task definition does not exist, taskDefinitionCode:{}.", taskCode); |
| putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode)); |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelations = |
| processTaskRelationMapper.queryByProcessCode(processDefinitionCode); |
| List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations); |
| if (CollectionUtils.isEmpty(processTaskRelationList)) { |
| log.error("Process task relations are empty, projectCode:{}, processDefinitionCode:{}.", projectCode, |
| processDefinitionCode); |
| putMsg(result, Status.DATA_IS_NULL, "processTaskRelationList"); |
| return result; |
| } |
| List<Long> downstreamList = Lists.newArrayList(); |
| for (ProcessTaskRelation processTaskRelation : processTaskRelations) { |
| if (processTaskRelation.getPreTaskCode() == taskCode) { |
| downstreamList.add(processTaskRelation.getPostTaskCode()); |
| } |
| if (processTaskRelation.getPostTaskCode() == taskCode) { |
| processTaskRelationList.remove(processTaskRelation); |
| } |
| } |
| if (CollectionUtils.isNotEmpty(downstreamList)) { |
| String downstream = StringUtils.join(downstreamList, ","); |
| log.warn( |
| "Relation can not be deleted because task has downstream tasks:[{}], projectCode:{}, processDefinitionCode:{}, taskDefinitionCode:{}.", |
| downstream, projectCode, processDefinitionCode, taskCode); |
| putMsg(result, Status.TASK_HAS_DOWNSTREAM, downstream); |
| return result; |
| } |
| updateProcessDefiniteVersion(loginUser, result, processDefinition); |
| updateRelation(loginUser, result, processDefinition, processTaskRelationList); |
| if (TASK_TYPE_CONDITIONS.equals(taskDefinition.getTaskType()) |
| || TASK_TYPE_DEPENDENT.equals(taskDefinition.getTaskType()) |
| || TASK_TYPE_SUB_PROCESS.equals(taskDefinition.getTaskType())) { |
| int deleteTaskDefinition = taskDefinitionMapper.deleteByCode(taskCode); |
| if (0 == deleteTaskDefinition) { |
| log.error("Delete task definition error, taskDefinitionCode:{}.", taskCode); |
| putMsg(result, Status.DELETE_TASK_DEFINE_BY_CODE_ERROR); |
| throw new ServiceException(Status.DELETE_TASK_DEFINE_BY_CODE_ERROR); |
| } else |
| log.info("Delete {} type task definition complete, taskDefinitionCode:{}.", |
| taskDefinition.getTaskType(), taskCode); |
| } |
| putMsg(result, Status.SUCCESS); |
| return result; |
| } |
| |
| /** |
| * delete process task relation, will delete exists relation preTaskCode -> postTaskCode, throw error if not exists |
| * |
| * @param loginUser login user |
| * @param preTaskCode relation upstream code |
| * @param postTaskCode relation downstream code |
| */ |
| @Override |
| @Transactional |
| public void deleteTaskProcessRelationV2(User loginUser, |
| long preTaskCode, |
| long postTaskCode) { |
| ProcessTaskRelation processTaskRelation = |
| new TaskRelationFilterRequest(preTaskCode, postTaskCode).convert2TaskDefinition(); |
| |
| Page<ProcessTaskRelation> page = |
| new Page<>(new TaskRelationFilterRequest(preTaskCode, postTaskCode).getPageNo(), |
| new TaskRelationFilterRequest(preTaskCode, postTaskCode).getPageSize()); |
| IPage<ProcessTaskRelation> processTaskRelationIPage = |
| processTaskRelationMapper.filterProcessTaskRelation(page, processTaskRelation); |
| |
| List<ProcessTaskRelation> processTaskRelations = processTaskRelationIPage.getRecords(); |
| if (processTaskRelations.size() != 1) { |
| throw new ServiceException(Status.PROCESS_TASK_RELATION_NOT_EXPECT, 1, processTaskRelations.size()); |
| } |
| |
| ProcessTaskRelation processTaskRelationDb = processTaskRelations.get(0); |
| Project project = projectMapper.queryByCode(processTaskRelationDb.getProjectCode()); |
| projectService.checkProjectAndAuthThrowException(loginUser, project, null); |
| processTaskRelationMapper.deleteById(processTaskRelationDb.getId()); |
| } |
| |
| /** |
| * delete process task relation, will delete exists relation upstream -> downstream, throw error if not exists |
| * |
| * @param loginUser login user |
| * @param taskCode relation upstream code |
| * @param needSyncDag needSyncDag |
| * @param taskRelationUpdateUpstreamRequest relation downstream code |
| */ |
| @Override |
| @Transactional |
| public List<ProcessTaskRelation> updateUpstreamTaskDefinitionWithSyncDag(User loginUser, |
| long taskCode, |
| Boolean needSyncDag, |
| TaskRelationUpdateUpstreamRequest taskRelationUpdateUpstreamRequest) { |
| TaskDefinition downstreamTask = taskDefinitionMapper.queryByCode(taskCode); |
| if (downstreamTask == null) { |
| throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, taskCode); |
| } |
| List<Long> upstreamTaskCodes = taskRelationUpdateUpstreamRequest.getUpstreams(); |
| |
| ProcessTaskRelation processTaskRelation = new ProcessTaskRelation(); |
| processTaskRelation.setPostTaskCode(taskCode); |
| |
| Page<ProcessTaskRelation> page = new Page<>(taskRelationUpdateUpstreamRequest.getPageNo(), |
| taskRelationUpdateUpstreamRequest.getPageSize()); |
| IPage<ProcessTaskRelation> processTaskRelationExistsIPage = |
| processTaskRelationMapper.filterProcessTaskRelation(page, processTaskRelation); |
| List<ProcessTaskRelation> processTaskRelationExists = processTaskRelationExistsIPage.getRecords(); |
| |
| ProcessDefinition processDefinition = null; |
| if (CollectionUtils.isNotEmpty(processTaskRelationExists)) { |
| processDefinition = |
| processDefinitionMapper.queryByCode(processTaskRelationExists.get(0).getProcessDefinitionCode()); |
| } else if (taskRelationUpdateUpstreamRequest.getWorkflowCode() != 0L) { |
| processDefinition = |
| processDefinitionMapper.queryByCode(taskRelationUpdateUpstreamRequest.getWorkflowCode()); |
| } |
| if (processDefinition == null) { |
| throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, |
| taskRelationUpdateUpstreamRequest.toString()); |
| } |
| processDefinition.setUpdateTime(new Date()); |
| int insertVersion = processDefinition.getVersion(); |
| if (needSyncDag) { |
| insertVersion = |
| this.saveProcessDefine(loginUser, processDefinition); |
| if (insertVersion <= 0) { |
| throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR); |
| } |
| } |
| // get new relation to create and out of date relation to delete |
| List<Long> taskCodeCreates = upstreamTaskCodes |
| .stream() |
| .filter(upstreamTaskCode -> processTaskRelationExists.stream().noneMatch( |
| processTaskRelationExist -> processTaskRelationExist.getPreTaskCode() == upstreamTaskCode)) |
| .collect(Collectors.toList()); |
| List<Integer> taskCodeDeletes = processTaskRelationExists.stream() |
| .filter(ptr -> !upstreamTaskCodes.contains(ptr.getPreTaskCode())) |
| .map(ProcessTaskRelation::getId) |
| .collect(Collectors.toList()); |
| |
| // delete relation not exists |
| if (CollectionUtils.isNotEmpty(taskCodeDeletes)) { |
| int delete = processTaskRelationMapper.deleteBatchIds(taskCodeDeletes); |
| if (delete != taskCodeDeletes.size()) { |
| throw new ServiceException(Status.PROCESS_TASK_RELATION_BATCH_DELETE_ERROR, taskCodeDeletes); |
| } |
| } |
| |
| // create relation not exists |
| List<ProcessTaskRelation> processTaskRelations = new ArrayList<>(); |
| for (long createCode : taskCodeCreates) { |
| long upstreamCode = 0L; |
| int version = 0; |
| if (createCode != 0L) { |
| // 0 for DAG root, should not, it may already exists and skip to create anymore |
| TaskDefinition upstreamTask = taskDefinitionMapper.queryByCode(createCode); |
| if (upstreamTask == null) { |
| throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, createCode); |
| } |
| upstreamCode = upstreamTask.getCode(); |
| version = upstreamTask.getVersion(); |
| } |
| ProcessTaskRelation processTaskRelationCreate = |
| new ProcessTaskRelation(null, processDefinition.getVersion(), downstreamTask.getProjectCode(), |
| processDefinition.getCode(), upstreamCode, version, |
| downstreamTask.getCode(), downstreamTask.getVersion(), null, null); |
| processTaskRelations.add(processTaskRelationCreate); |
| } |
| int batchInsert = processTaskRelationMapper.batchInsert(processTaskRelations); |
| if (batchInsert != processTaskRelations.size()) { |
| throw new ServiceException(Status.PROCESS_TASK_RELATION_BATCH_CREATE_ERROR, taskCodeCreates); |
| } |
| |
| // batch sync to process task relation log |
| int saveTaskRelationResult = saveTaskRelation(loginUser, processDefinition, insertVersion); |
| if (saveTaskRelationResult != Constants.EXIT_CODE_SUCCESS) { |
| log.error("Save process task relations error, projectCode:{}, processCode:{}, processVersion:{}.", |
| processDefinition.getProjectCode(), processDefinition.getCode(), insertVersion); |
| throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR); |
| } |
| log.info("Save process task relations complete, projectCode:{}, processCode:{}, processVersion:{}.", |
| processDefinition.getProjectCode(), processDefinition.getCode(), insertVersion); |
| processTaskRelations.get(0).setProcessDefinitionVersion(insertVersion); |
| return processTaskRelations; |
| } |
| |
| public int saveTaskRelation(User loginUser, ProcessDefinition processDefinition, |
| int processDefinitionVersion) { |
| long projectCode = processDefinition.getProjectCode(); |
| long processDefinitionCode = processDefinition.getCode(); |
| List<ProcessTaskRelation> taskRelations = |
| processTaskRelationMapper.queryByProcessCode(processDefinitionCode); |
| List<ProcessTaskRelationLog> taskRelationList = |
| taskRelations.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList()); |
| |
| List<Long> taskCodeList = |
| taskRelations.stream().map(ProcessTaskRelation::getPostTaskCode).collect(Collectors.toList()); |
| List<TaskDefinition> taskDefinitions = taskDefinitionMapper.queryByCodeList(taskCodeList); |
| List<TaskDefinitionLog> taskDefinitionLogs = |
| taskDefinitions.stream().map(TaskDefinitionLog::new).collect(Collectors.toList()); |
| |
| if (taskRelationList.isEmpty()) { |
| return Constants.EXIT_CODE_SUCCESS; |
| } |
| Map<Long, TaskDefinitionLog> taskDefinitionLogMap = null; |
| if (org.apache.commons.collections.CollectionUtils.isNotEmpty(taskDefinitionLogs)) { |
| taskDefinitionLogMap = taskDefinitionLogs |
| .stream() |
| .collect(Collectors.toMap(TaskDefinition::getCode, taskDefinitionLog -> taskDefinitionLog)); |
| } |
| Date now = new Date(); |
| for (ProcessTaskRelationLog processTaskRelationLog : taskRelationList) { |
| processTaskRelationLog.setProjectCode(projectCode); |
| processTaskRelationLog.setProcessDefinitionCode(processDefinitionCode); |
| processTaskRelationLog.setProcessDefinitionVersion(processDefinitionVersion); |
| if (taskDefinitionLogMap != null) { |
| TaskDefinitionLog preTaskDefinitionLog = |
| taskDefinitionLogMap.get(processTaskRelationLog.getPreTaskCode()); |
| if (preTaskDefinitionLog != null) { |
| processTaskRelationLog.setPreTaskVersion(preTaskDefinitionLog.getVersion()); |
| } |
| TaskDefinitionLog postTaskDefinitionLog = |
| taskDefinitionLogMap.get(processTaskRelationLog.getPostTaskCode()); |
| if (postTaskDefinitionLog != null) { |
| processTaskRelationLog.setPostTaskVersion(postTaskDefinitionLog.getVersion()); |
| } |
| } |
| processTaskRelationLog.setCreateTime(now); |
| processTaskRelationLog.setUpdateTime(now); |
| processTaskRelationLog.setOperator(loginUser.getId()); |
| processTaskRelationLog.setOperateTime(now); |
| } |
| if (CollectionUtils.isNotEmpty(taskRelations)) { |
| Set<Integer> processTaskRelationSet = |
| taskRelations.stream().map(ProcessTaskRelation::hashCode).collect(toSet()); |
| Set<Integer> taskRelationSet = |
| taskRelationList.stream().map(ProcessTaskRelationLog::hashCode).collect(toSet()); |
| boolean isSame = org.apache.commons.collections.CollectionUtils.isEqualCollection(processTaskRelationSet, |
| taskRelationSet); |
| if (isSame) { |
| return Constants.EXIT_CODE_SUCCESS; |
| } |
| processTaskRelationMapper.deleteByCode(projectCode, processDefinitionCode); |
| } |
| List<ProcessTaskRelation> processTaskRelations = |
| taskRelationList.stream().map(ProcessTaskRelation::new).collect(Collectors.toList()); |
| int insert = processTaskRelationMapper.batchInsert(processTaskRelations); |
| int resultLog = processTaskRelationLogMapper.batchInsert(taskRelationList); |
| return (insert & resultLog) > 0 ? Constants.EXIT_CODE_SUCCESS : Constants.EXIT_CODE_FAILURE; |
| } |
| |
| public int saveProcessDefine(User loginUser, ProcessDefinition processDefinition) { |
| ProcessDefinitionLog processDefinitionLog = new ProcessDefinitionLog(processDefinition); |
| Integer version = processDefinitionLogMapper.queryMaxVersionForDefinition(processDefinition.getCode()); |
| int insertVersion = version == null || version == 0 ? Constants.VERSION_FIRST : version + 1; |
| processDefinitionLog.setVersion(insertVersion); |
| processDefinitionLog.setOperator(loginUser.getId()); |
| processDefinitionLog.setOperateTime(processDefinition.getUpdateTime()); |
| processDefinitionLog.setId(null); |
| int insertLog = processDefinitionLogMapper.insert(processDefinitionLog); |
| |
| processDefinitionLog.setId(processDefinition.getId()); |
| int result = processDefinitionMapper.updateById(processDefinitionLog); |
| return (insertLog & result) > 0 ? insertVersion : 0; |
| } |
| |
| private void updateRelation(User loginUser, Map<String, Object> result, ProcessDefinition processDefinition, |
| List<ProcessTaskRelation> processTaskRelationList) { |
| List<ProcessTaskRelationLog> relationLogs = |
| processTaskRelationList.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList()); |
| int insertResult = processService.saveTaskRelation(loginUser, processDefinition.getProjectCode(), |
| processDefinition.getCode(), |
| processDefinition.getVersion(), relationLogs, Lists.newArrayList(), Boolean.TRUE); |
| if (insertResult == Constants.EXIT_CODE_SUCCESS) { |
| log.info( |
| "Update task relations complete, projectCode:{}, processDefinitionCode:{}, processDefinitionVersion:{}.", |
| processDefinition.getProjectCode(), processDefinition.getCode(), processDefinition.getVersion()); |
| putMsg(result, Status.SUCCESS); |
| result.put(Constants.DATA_LIST, processDefinition); |
| } else { |
| log.error( |
| "Update task relations error, projectCode:{}, processDefinitionCode:{}, processDefinitionVersion:{}.", |
| processDefinition.getProjectCode(), processDefinition.getCode(), processDefinition.getVersion()); |
| putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR); |
| throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR); |
| } |
| } |
| |
| /** |
| * delete task upstream relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param preTaskCodes the pre task codes, sep ',' |
| * @param taskCode the post task code |
| * @return delete result code |
| */ |
| @Transactional |
| @Override |
| public Map<String, Object> deleteUpstreamRelation(User loginUser, long projectCode, String preTaskCodes, |
| long taskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| if (StringUtils.isEmpty(preTaskCodes)) { |
| log.warn("Parameter preTaskCodes is empty."); |
| putMsg(result, Status.DATA_IS_NULL, "preTaskCodes"); |
| return result; |
| } |
| List<ProcessTaskRelation> upstreamList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); |
| if (CollectionUtils.isEmpty(upstreamList)) { |
| log.error("Upstream tasks based on the task do not exist, theTaskDefinitionCode:{}.", taskCode); |
| putMsg(result, Status.DATA_IS_NULL, "taskCode"); |
| return result; |
| } |
| |
| List<Long> preTaskCodeList = Lists.newArrayList(preTaskCodes.split(Constants.COMMA)).stream() |
| .map(Long::parseLong).collect(Collectors.toList()); |
| if (preTaskCodeList.contains(0L)) { |
| log.warn("Parameter preTaskCodes contain 0."); |
| putMsg(result, Status.DATA_IS_NULL, "preTaskCodes"); |
| return result; |
| } |
| List<Long> currentUpstreamList = |
| upstreamList.stream().map(ProcessTaskRelation::getPreTaskCode).collect(Collectors.toList()); |
| if (currentUpstreamList.contains(0L)) { |
| log.error("Upstream taskCodes based on the task contain, theTaskDefinitionCode:{}.", taskCode); |
| putMsg(result, Status.DATA_IS_NOT_VALID, "currentUpstreamList"); |
| return result; |
| } |
| List<Long> tmpCurrent = Lists.newArrayList(currentUpstreamList); |
| tmpCurrent.removeAll(preTaskCodeList); |
| preTaskCodeList.removeAll(currentUpstreamList); |
| if (!preTaskCodeList.isEmpty()) { |
| String invalidPreTaskCodes = StringUtils.join(preTaskCodeList, Constants.COMMA); |
| log.error("Some upstream taskCodes are invalid, preTaskCodeList:{}.", invalidPreTaskCodes); |
| putMsg(result, Status.DATA_IS_NOT_VALID, invalidPreTaskCodes); |
| return result; |
| } |
| ProcessDefinition processDefinition = |
| processDefinitionMapper.queryByCode(upstreamList.get(0).getProcessDefinitionCode()); |
| if (processDefinition == null) { |
| log.error("Process definition does not exist, processDefinitionCode:{}.", |
| upstreamList.get(0).getProcessDefinitionCode()); |
| putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, |
| String.valueOf(upstreamList.get(0).getProcessDefinitionCode())); |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelations = |
| processTaskRelationMapper.queryByProcessCode(processDefinition.getCode()); |
| List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations); |
| List<ProcessTaskRelation> processTaskRelationWaitRemove = Lists.newArrayList(); |
| for (ProcessTaskRelation processTaskRelation : processTaskRelationList) { |
| if (currentUpstreamList.size() > 1) { |
| if (currentUpstreamList.contains(processTaskRelation.getPreTaskCode())) { |
| currentUpstreamList.remove(processTaskRelation.getPreTaskCode()); |
| processTaskRelationWaitRemove.add(processTaskRelation); |
| } |
| } else { |
| if (processTaskRelation.getPostTaskCode() == taskCode |
| && (currentUpstreamList.isEmpty() || tmpCurrent.isEmpty())) { |
| processTaskRelation.setPreTaskVersion(0); |
| processTaskRelation.setPreTaskCode(0L); |
| } |
| } |
| } |
| processTaskRelationList.removeAll(processTaskRelationWaitRemove); |
| updateProcessDefiniteVersion(loginUser, result, processDefinition); |
| updateRelation(loginUser, result, processDefinition, processTaskRelationList); |
| return result; |
| } |
| |
| /** |
| * delete task downstream relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param postTaskCodes the post task codes, sep ',' |
| * @param taskCode the pre task code |
| * @return delete result code |
| */ |
| @Transactional |
| @Override |
| public Map<String, Object> deleteDownstreamRelation(User loginUser, long projectCode, String postTaskCodes, |
| long taskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| if (StringUtils.isEmpty(postTaskCodes)) { |
| log.warn("Parameter postTaskCodes is empty."); |
| putMsg(result, Status.DATA_IS_NULL, "postTaskCodes"); |
| return result; |
| } |
| List<ProcessTaskRelation> downstreamList = |
| processTaskRelationMapper.queryDownstreamByCode(projectCode, taskCode); |
| if (CollectionUtils.isEmpty(downstreamList)) { |
| log.error("Downstream tasks based on the task do not exist, theTaskDefinitionCode:{}.", taskCode); |
| putMsg(result, Status.DATA_IS_NULL, "taskCode"); |
| return result; |
| } |
| List<Long> postTaskCodeList = Lists.newArrayList(postTaskCodes.split(Constants.COMMA)).stream() |
| .map(Long::parseLong).collect(Collectors.toList()); |
| if (postTaskCodeList.contains(0L)) { |
| log.warn("Parameter postTaskCodes contains 0."); |
| putMsg(result, Status.DATA_IS_NULL, "postTaskCodes"); |
| return result; |
| } |
| ProcessDefinition processDefinition = |
| processDefinitionMapper.queryByCode(downstreamList.get(0).getProcessDefinitionCode()); |
| if (processDefinition == null) { |
| log.error("Process definition does not exist, processDefinitionCode:{}.", |
| downstreamList.get(0).getProcessDefinitionCode()); |
| putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, |
| String.valueOf(downstreamList.get(0).getProcessDefinitionCode())); |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelations = |
| processTaskRelationMapper.queryByProcessCode(processDefinition.getCode()); |
| List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations); |
| processTaskRelationList |
| .removeIf(processTaskRelation -> postTaskCodeList.contains(processTaskRelation.getPostTaskCode()) |
| && processTaskRelation.getPreTaskCode() == taskCode); |
| updateProcessDefiniteVersion(loginUser, result, processDefinition); |
| updateRelation(loginUser, result, processDefinition, processTaskRelationList); |
| return result; |
| } |
| |
| /** |
| * query task upstream relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param taskCode current task code (post task code) |
| * @return the upstream task definitions |
| */ |
| @Override |
| public Map<String, Object> queryUpstreamRelation(User loginUser, long projectCode, long taskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelationList = |
| processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); |
| List<TaskDefinitionLog> taskDefinitionLogList = new ArrayList<>(); |
| if (CollectionUtils.isNotEmpty(processTaskRelationList)) { |
| Set<TaskDefinition> taskDefinitions = processTaskRelationList |
| .stream() |
| .map(processTaskRelation -> { |
| TaskDefinition taskDefinition = buildTaskDefinition(); |
| taskDefinition.setProjectCode(processTaskRelation.getProjectCode()); |
| taskDefinition.setCode(processTaskRelation.getPreTaskCode()); |
| taskDefinition.setVersion(processTaskRelation.getPreTaskVersion()); |
| return taskDefinition; |
| }) |
| .collect(Collectors.toSet()); |
| taskDefinitionLogList = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitions); |
| } |
| result.put(Constants.DATA_LIST, taskDefinitionLogList); |
| putMsg(result, Status.SUCCESS); |
| return result; |
| } |
| |
| /** |
| * query task downstream relation |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param taskCode pre task code |
| * @return the downstream task definitions |
| */ |
| @Override |
| public Map<String, Object> queryDownstreamRelation(User loginUser, long projectCode, long taskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelationList = |
| processTaskRelationMapper.queryDownstreamByCode(projectCode, taskCode); |
| List<TaskDefinitionLog> taskDefinitionLogList = new ArrayList<>(); |
| if (CollectionUtils.isNotEmpty(processTaskRelationList)) { |
| Set<TaskDefinition> taskDefinitions = processTaskRelationList |
| .stream() |
| .map(processTaskRelation -> { |
| TaskDefinition taskDefinition = buildTaskDefinition(); |
| taskDefinition.setProjectCode(processTaskRelation.getProjectCode()); |
| taskDefinition.setCode(processTaskRelation.getPostTaskCode()); |
| taskDefinition.setVersion(processTaskRelation.getPostTaskVersion()); |
| return taskDefinition; |
| }) |
| .collect(Collectors.toSet()); |
| taskDefinitionLogList = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitions); |
| } |
| result.put(Constants.DATA_LIST, taskDefinitionLogList); |
| putMsg(result, Status.SUCCESS); |
| return result; |
| } |
| |
| /** |
| * delete edge |
| * |
| * @param loginUser login user |
| * @param projectCode project code |
| * @param processDefinitionCode process definition code |
| * @param preTaskCode pre task code |
| * @param postTaskCode post task code |
| * @return delete result code |
| */ |
| @Transactional |
| @Override |
| public Map<String, Object> deleteEdge(User loginUser, long projectCode, long processDefinitionCode, |
| long preTaskCode, long postTaskCode) { |
| Project project = projectMapper.queryByCode(projectCode); |
| // check user access for project |
| Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null); |
| if (result.get(Constants.STATUS) != Status.SUCCESS) { |
| return result; |
| } |
| ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode); |
| if (processDefinition == null) { |
| log.error("Process definition does not exist, projectCode：{}， processDefinitionCode:{}.", projectCode, |
| processDefinitionCode); |
| putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode)); |
| return result; |
| } |
| List<ProcessTaskRelation> processTaskRelations = |
| processTaskRelationMapper.queryByProcessCode(processDefinitionCode); |
| List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations); |
| if (CollectionUtils.isEmpty(processTaskRelationList)) { |
| log.error("Process task relations are empty, projectCode:{}, processDefinitionCode:{}.", projectCode, |
| processDefinitionCode); |
| putMsg(result, Status.DATA_IS_NULL, "processTaskRelationList"); |
| return result; |
| } |
| Map<Long, List<ProcessTaskRelation>> taskRelationMap = new HashMap<>(); |
| for (ProcessTaskRelation processTaskRelation : processTaskRelationList) { |
| taskRelationMap.compute(processTaskRelation.getPostTaskCode(), (k, v) -> { |
| if (v == null) { |
| v = new ArrayList<>(); |
| } |
| v.add(processTaskRelation); |
| return v; |
| }); |
| } |
| if (!taskRelationMap.containsKey(postTaskCode)) { |
| putMsg(result, Status.DATA_IS_NULL, "postTaskCode"); |
| return result; |
| } |
| if (taskRelationMap.get(postTaskCode).size() > 1) { |
| for (ProcessTaskRelation processTaskRelation : taskRelationMap.get(postTaskCode)) { |
| if (processTaskRelation.getPreTaskCode() == preTaskCode) { |
| int delete = processTaskRelationMapper.deleteById(processTaskRelation.getId()); |
| if (delete == 0) { |
| log.error( |
| "Delete task relation edge error, processTaskRelationId:{}, preTaskCode:{}, postTaskCode:{}", |
| processTaskRelation.getId(), preTaskCode, postTaskCode); |
| putMsg(result, Status.DELETE_EDGE_ERROR); |
| throw new ServiceException(Status.DELETE_EDGE_ERROR); |
| } else |
| log.info( |
| "Delete task relation edge complete, processTaskRelationId:{}, preTaskCode:{}, postTaskCode:{}", |
| processTaskRelation.getId(), preTaskCode, postTaskCode); |
| processTaskRelationList.remove(processTaskRelation); |
| } |
| } |
| } else { |
| ProcessTaskRelation processTaskRelation = taskRelationMap.get(postTaskCode).get(0); |
| processTaskRelationList.remove(processTaskRelation); |
| processTaskRelation.setPreTaskVersion(0); |
| processTaskRelation.setPreTaskCode(0L); |
| processTaskRelationList.add(processTaskRelation); |
| log.info( |
| "Delete task relation through set invalid value for it: preTaskCode from {} to 0, processTaskRelationId:{}.", |
| preTaskCode, processTaskRelation.getId()); |
| } |
| updateProcessDefiniteVersion(loginUser, result, processDefinition); |
| updateRelation(loginUser, result, processDefinition, processTaskRelationList); |
| return result; |
| } |
| |
| @Override |
| public List<ProcessTaskRelation> queryByWorkflowDefinitionCode(long workflowDefinitionCode, |
| int workflowDefinitionVersion) { |
| return processTaskRelationMapper.queryProcessTaskRelationsByProcessDefinitionCode(workflowDefinitionCode, |
| workflowDefinitionVersion); |
| } |
| |
| @Override |
| public void deleteByWorkflowDefinitionCode(long workflowDefinitionCode, int workflowDefinitionVersion) { |
| processTaskRelationMapper.deleteByWorkflowDefinitionCode(workflowDefinitionCode, workflowDefinitionVersion); |
| } |
| |
| /** |
| * build task definition |
| * |
| * @return task definition |
| */ |
| private TaskDefinition buildTaskDefinition() { |
| |
| return new TaskDefinition() { |
| |
| @Override |
| public boolean equals(Object o) { |
| if (this == o) { |
| return true; |
| } |
| if (!(o instanceof TaskDefinition)) { |
| return false; |
| } |
| TaskDefinition that = (TaskDefinition) o; |
| return getCode() == that.getCode() |
| && getVersion() == that.getVersion() |
| && getProjectCode() == that.getProjectCode(); |
| } |
| |
| @Override |
| public int hashCode() { |
| return Objects.hash(getCode(), getVersion(), getProjectCode()); |
| } |
| }; |
| } |
| } |
