gtsoft/gtsoft-snail-job-server
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat(sj_1.0.0): 新增组导入、导出接口

Browse Source
This commit is contained in:
opensnail 2024-05-27 11:37:53 +08:00
parent 8d46c8d181
commit d72455fb1a
7 changed files with 217 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/controller/GroupConfigController.java
View File

@ -1,19 +1,36 @@
package com.aizuda.snailjob.server.web.controller;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.lang.Assert;
import com.aizuda.snailjob.common.core.annotation.OriginalControllerReturnValue;
import com.aizuda.snailjob.common.core.exception.SnailJobCommonException;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.server.common.exception.SnailJobServerException;
import com.aizuda.snailjob.server.common.util.DateUtils;
import com.aizuda.snailjob.server.web.annotation.LoginRequired;
import com.aizuda.snailjob.server.web.annotation.LoginUser;
import com.aizuda.snailjob.server.web.annotation.RoleEnum;
import com.aizuda.snailjob.server.web.model.base.PageResult;
import com.aizuda.snailjob.server.web.model.request.GroupConfigQueryVO;
import com.aizuda.snailjob.server.web.model.request.GroupConfigRequestVO;
import com.aizuda.snailjob.server.web.model.request.UserSessionVO;
import com.aizuda.snailjob.server.web.model.response.GroupConfigResponseVO;
import com.aizuda.snailjob.server.web.service.GroupConfigService;
import org.springframework.beans.factory.annotation.Autowired;
import com.fasterxml.jackson.databind.JsonNode;
import jakarta.validation.ConstraintViolation;
import jakarta.validation.Validation;
import jakarta.validation.Validator;
import jakarta.validation.ValidatorFactory;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Set;
/**
* 重试组接口
@ -23,15 +40,15 @@ import java.util.List;
*/
@RestController
@RequestMapping("/group")
@RequiredArgsConstructor
public class GroupConfigController {
@Autowired
private GroupConfigService groupConfigService;
private static final List<String> FILE_EXTENSIONS = List.of("json");
private final GroupConfigService groupConfigService;
@LoginRequired(role = RoleEnum.ADMIN)
@PostMapping("")
public Boolean addGroup(@LoginUser UserSessionVO systemUser, @RequestBody @Validated GroupConfigRequestVO groupConfigRequestVO) {
return groupConfigService.addGroup(systemUser, groupConfigRequestVO);
public Boolean addGroup(@RequestBody @Validated GroupConfigRequestVO groupConfigRequestVO) {
return groupConfigService.addGroup(groupConfigRequestVO);
}
@LoginRequired(role = RoleEnum.ADMIN)
@ -84,4 +101,54 @@ public class GroupConfigController {
return groupConfigService.getTablePartitionList();
}
@PostMapping("/import")
@LoginRequired
public void importScene(@RequestParam("file") MultipartFile file) throws IOException {
if (file.isEmpty()) {
throw new SnailJobCommonException("Please select a file to upload");
}
// 保存文件到服务器
String suffix = FileUtil.getSuffix(file.getOriginalFilename());
if (!FILE_EXTENSIONS.contains(suffix)) {
throw new SnailJobCommonException("文件类型错误");
}
JsonNode node = JsonUtil.toJson(file.getBytes());
List<GroupConfigRequestVO> requestList = JsonUtil.parseList(JsonUtil.toJsonString(node),
GroupConfigRequestVO.class);
// 校验参数是否合法
for (final GroupConfigRequestVO groupConfigRequestVO : requestList) {
ValidatorFactory vf = Validation.buildDefaultValidatorFactory();
Validator validator = vf.getValidator();
Set<ConstraintViolation<GroupConfigRequestVO>> set = validator.validate(groupConfigRequestVO);
for (final ConstraintViolation<GroupConfigRequestVO> violation : set) {
throw new SnailJobCommonException(violation.getMessage());
}
}
groupConfigService.importGroup(requestList);
}
@PostMapping("/export")
@LoginRequired
@OriginalControllerReturnValue
public ResponseEntity<String> exportGroup(@RequestBody Set<Long> groupIds) {
Assert.notEmpty(groupIds, () -> new SnailJobServerException("参数错误"));
String configs = groupConfigService.exportGroup(groupIds);
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
// 设置下载时的文件名称
String fileName = String.format("group-config-%s.json", DateUtils.toNowFormat(DateUtils.PURE_DATETIME_MS_PATTERN));
String disposition = "attachment; filename=" +
new String(fileName.getBytes(StandardCharsets.UTF_8), StandardCharsets.ISO_8859_1);
headers.add(HttpHeaders.CONTENT_DISPOSITION, disposition);
return ResponseEntity.ok()
.headers(headers)
.body(configs);
}
}

18
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/controller/SceneConfigController.java
View File

@ -1,11 +1,8 @@
package com.aizuda.snailjob.server.web.controller;
import cn.hutool.core.io.FileUtil;
import cn.hutool.http.server.HttpServerResponse;
import com.aizuda.snailjob.client.model.request.DispatchJobRequest;
import com.aizuda.snailjob.common.core.annotation.OriginalControllerReturnValue;
import com.aizuda.snailjob.common.core.exception.SnailJobCommonException;
import com.aizuda.snailjob.common.core.model.Result;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.server.common.util.DateUtils;
import com.aizuda.snailjob.server.web.annotation.LoginRequired;
@ -20,7 +17,7 @@ import jakarta.validation.Validation;
import jakarta.validation.Validator;
import jakarta.validation.ValidatorFactory;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.UrlResource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
@ -30,7 +27,6 @@ import org.springframework.web.multipart.MultipartFile;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
@ -45,7 +41,7 @@ import java.util.Set;
@RequiredArgsConstructor
public class SceneConfigController {
private final List<String> FILE_EXTENSIONS = List.of("json");
private static final List<String> FILE_EXTENSIONS = List.of("json");
private final SceneConfigService sceneConfigService;
@LoginRequired
@ -86,8 +82,7 @@ public class SceneConfigController {
@PostMapping("/import")
@LoginRequired
public void importScene(final MultipartFile file)
throws IOException {
public void importScene(final MultipartFile file) throws IOException {
if (file.isEmpty()) {
throw new SnailJobCommonException("Please select a file to upload");
}
@ -126,7 +121,7 @@ public class SceneConfigController {
headers.setContentType(MediaType.APPLICATION_JSON);
// 设置下载时的文件名称
String fileName = String.format("%s.json", DateUtils.toNowFormat(DateUtils.PURE_DATETIME_MS_PATTERN));
String fileName = String.format("retry-scene-%s.json", DateUtils.toNowFormat(DateUtils.PURE_DATETIME_MS_PATTERN));
String disposition = "attachment; filename=" +
new String(fileName.getBytes(StandardCharsets.UTF_8), StandardCharsets.ISO_8859_1);
headers.add(HttpHeaders.CONTENT_DISPOSITION, disposition);
@ -135,9 +130,4 @@ public class SceneConfigController {
.body(configs);
}
@LoginRequired
@PostMapping("/{targetNamespaceId}/batch/copy")
public void batchCopy(@PathVariable("targetNamespaceId") Long targetNamespaceId, @RequestBody Set<Long> sceneIds) {
sceneConfigService.batchCopy(targetNamespaceId, sceneIds);
}
}

8
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/service/GroupConfigService.java
View File

@ -3,10 +3,10 @@ package com.aizuda.snailjob.server.web.service;
import com.aizuda.snailjob.server.web.model.base.PageResult;
import com.aizuda.snailjob.server.web.model.request.GroupConfigQueryVO;
import com.aizuda.snailjob.server.web.model.request.GroupConfigRequestVO;
import com.aizuda.snailjob.server.web.model.request.UserSessionVO;
import com.aizuda.snailjob.server.web.model.response.GroupConfigResponseVO;
import java.util.List;
import java.util.Set;
/**
* @author: opensnail
@ -14,7 +14,7 @@ import java.util.List;
*/
public interface GroupConfigService {
Boolean addGroup(UserSessionVO systemUser, GroupConfigRequestVO groupConfigRequestVO);
Boolean addGroup(GroupConfigRequestVO groupConfigRequestVO);
Boolean updateGroup(GroupConfigRequestVO groupConfigRequestVO);
@ -31,4 +31,8 @@ public interface GroupConfigService {
List<String> getOnlinePods(String groupName);
List<Integer> getTablePartitionList();
void importGroup(List<GroupConfigRequestVO> requestVOS);
String exportGroup(Set<Long> groupIds);
}

1
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/service/SceneConfigService.java
View File

@ -30,5 +30,4 @@ public interface SceneConfigService {
String exportSceneConfig(Set<Long> sceneIds);
void batchCopy(Long targetNamespaceId, Set<Long> sceneIds);
}

8
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/service/convert/GroupConfigConverter.java
View File

@ -18,7 +18,11 @@ public interface GroupConfigConverter {
GroupConfigConverter INSTANCE = Mappers.getMapper(GroupConfigConverter.class);
GroupConfig convert(GroupConfigRequestVO groupConfigRequestVO);
GroupConfig toGroupConfig(GroupConfigRequestVO groupConfigRequestVO);
List<GroupConfig> toGroupConfigs(List<GroupConfigRequestVO> groupConfigRequestVOS);
List<GroupConfigRequestVO> toGroupConfigRequestVOs(List<GroupConfig> groupConfigs);
List<GroupConfig> convertList(List<GroupConfigRequestVO> groupConfigRequestVOS);
}

183
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/service/impl/GroupConfigServiceImpl.java
View File

@ -5,6 +5,7 @@ import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.HashUtil;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.common.core.util.StreamUtils;
import com.aizuda.snailjob.server.common.config.SystemProperties;
import com.aizuda.snailjob.server.common.enums.IdGeneratorModeEnum;
@ -32,8 +33,9 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.collect.Sets.SetView;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Lazy;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
@ -63,7 +65,6 @@ public class GroupConfigServiceImpl implements GroupConfigService {
private final ServerNodeMapper serverNodeMapper;
private final AccessTemplate accessTemplate;
private final SequenceAllocMapper sequenceAllocMapper;
@Lazy
private final ConfigVersionSyncHandler configVersionSyncHandler;
private final SystemProperties systemProperties;
private final JdbcTemplate jdbcTemplate;
@ -71,19 +72,21 @@ public class GroupConfigServiceImpl implements GroupConfigService {
@Override
@Transactional
public Boolean addGroup(UserSessionVO systemUser, GroupConfigRequestVO groupConfigRequestVO) {
public Boolean addGroup(GroupConfigRequestVO groupConfigRequestVO) {
String namespaceId = UserSessionUtils.currentUserSession().getNamespaceId();
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
Assert.isTrue(groupConfigAccess.count(new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, systemUser.getNamespaceId())
.eq(GroupConfig::getGroupName, groupConfigRequestVO.getGroupName())) == 0,
() -> new SnailJobServerException("GroupName已经存在 {}", groupConfigRequestVO.getGroupName()));
.eq(GroupConfig::getNamespaceId, namespaceId)
.eq(GroupConfig::getGroupName, groupConfigRequestVO.getGroupName())) == 0,
() -> new SnailJobServerException("GroupName已经存在 {}", groupConfigRequestVO.getGroupName()));
// 保存组配置
Boolean isSuccess = doSaveGroupConfig(systemUser, groupConfigRequestVO);
Boolean isSuccess = doSaveGroupConfig(namespaceId, groupConfigRequestVO);
// 保存生成唯一id配置
doSaveSequenceAlloc(systemUser, groupConfigRequestVO);
doSaveSequenceAlloc(namespaceId, groupConfigRequestVO);
return isSuccess;
}
@ -91,18 +94,18 @@ public class GroupConfigServiceImpl implements GroupConfigService {
/**
* 保存序号生成规则配置失败
*
* @param systemUser
* @param namespaceId 命名空间
* @param groupConfigRequestVO 组、场景、通知配置类
*/
private void doSaveSequenceAlloc(UserSessionVO systemUser, final GroupConfigRequestVO groupConfigRequestVO) {
private void doSaveSequenceAlloc(final String namespaceId, final GroupConfigRequestVO groupConfigRequestVO) {
SequenceAlloc sequenceAlloc = new SequenceAlloc();
sequenceAlloc.setGroupName(groupConfigRequestVO.getGroupName());
sequenceAlloc.setNamespaceId(systemUser.getNamespaceId());
sequenceAlloc.setNamespaceId(namespaceId);
sequenceAlloc.setStep(systemProperties.getStep());
sequenceAlloc.setUpdateDt(LocalDateTime.now());
Assert.isTrue(1 == sequenceAllocMapper.insert(sequenceAlloc),
() -> new SnailJobServerException("failed to save sequence generation rule configuration [{}].",
groupConfigRequestVO.getGroupName()));
() -> new SnailJobServerException("failed to save sequence generation rule configuration [{}].",
groupConfigRequestVO.getGroupName()));
}
@Override
@ -119,32 +122,32 @@ public class GroupConfigServiceImpl implements GroupConfigService {
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
long count = groupConfigAccess.count(
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.eq(GroupConfig::getGroupName, groupName));
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.eq(GroupConfig::getGroupName, groupName));
if (count <= 0) {
return false;
}
GroupConfig groupConfig = GroupConfigConverter.INSTANCE.convert(groupConfigRequestVO);
GroupConfig groupConfig = GroupConfigConverter.INSTANCE.toGroupConfig(groupConfigRequestVO);
groupConfig.setDescription(Optional.ofNullable(groupConfigRequestVO.getDescription()).orElse(StrUtil.EMPTY));
// 使用@TableField(value = "version", update= "%s+1") 进行更新version, 这里必须初始化一个值
groupConfig.setVersion(1);
groupConfig.setToken(null);
Assert.isTrue(tablePartitionList.contains(groupConfigRequestVO.getGroupPartition()),
() -> new SnailJobServerException("分区不存在. [{}]", tablePartitionList));
() -> new SnailJobServerException("分区不存在. [{}]", tablePartitionList));
Assert.isTrue(groupConfigRequestVO.getGroupPartition() >= 0,
() -> new SnailJobServerException("分区不能是负数."));
() -> new SnailJobServerException("分区不能是负数."));
// 校验retry_task_x和retry_dead_letter_x是否存在
checkGroupPartition(groupConfig, namespaceId);
Assert.isTrue(1 == groupConfigAccess.update(groupConfig,
new LambdaUpdateWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.eq(GroupConfig::getGroupName, groupName)),
() -> new SnailJobServerException("exception occurred while adding group. groupConfigVO[{}]",
groupConfigRequestVO));
new LambdaUpdateWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.eq(GroupConfig::getGroupName, groupName)),
() -> new SnailJobServerException("exception occurred while adding group. groupConfigVO[{}]",
groupConfigRequestVO));
// 同步版本， 版本为0代表需要同步到客户端
boolean add = configVersionSyncHandler.addSyncTask(groupName, namespaceId, 0);
@ -161,9 +164,9 @@ public class GroupConfigServiceImpl implements GroupConfigService {
groupConfig.setGroupStatus(status);
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
return groupConfigAccess.update(groupConfig,
new LambdaUpdateWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(GroupConfig::getGroupName, groupName)) == 1;
new LambdaUpdateWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(GroupConfig::getGroupName, groupName)) == 1;
}
@Override
@ -174,29 +177,30 @@ public class GroupConfigServiceImpl implements GroupConfigService {
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
PageDTO<GroupConfig> groupConfigPageDTO = groupConfigAccess.listPage(
new PageDTO<>(queryVO.getPage(), queryVO.getSize()),
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.in(userSessionVO.isUser(), GroupConfig::getGroupName, userSessionVO.getGroupNames())
.likeRight(StrUtil.isNotBlank(queryVO.getGroupName()), GroupConfig::getGroupName, StrUtil.trim(queryVO.getGroupName()))
.orderByDesc(GroupConfig::getId));
new PageDTO<>(queryVO.getPage(), queryVO.getSize()),
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.in(userSessionVO.isUser(), GroupConfig::getGroupName, userSessionVO.getGroupNames())
.likeRight(StrUtil.isNotBlank(queryVO.getGroupName()), GroupConfig::getGroupName,
StrUtil.trim(queryVO.getGroupName()))
.orderByDesc(GroupConfig::getId));
List<GroupConfig> records = groupConfigPageDTO.getRecords();
if (CollectionUtils.isEmpty(records)) {
return new PageResult<>(groupConfigPageDTO.getCurrent(), groupConfigPageDTO.getSize(),
groupConfigPageDTO.getTotal());
groupConfigPageDTO.getTotal());
}
PageResult<List<GroupConfigResponseVO>> pageResult = new PageResult<>(groupConfigPageDTO.getCurrent(),
groupConfigPageDTO.getSize(), groupConfigPageDTO.getTotal());
groupConfigPageDTO.getSize(), groupConfigPageDTO.getTotal());
List<GroupConfigResponseVO> responseVOList = GroupConfigResponseVOConverter.INSTANCE.convertList(
records);
records);
for (GroupConfigResponseVO groupConfigResponseVO : responseVOList) {
Optional.ofNullable(IdGeneratorModeEnum.modeOf(groupConfigResponseVO.getIdGeneratorMode()))
.ifPresent(idGeneratorMode -> {
groupConfigResponseVO.setIdGeneratorModeName(idGeneratorMode.getDesc());
});
.ifPresent(idGeneratorMode -> {
groupConfigResponseVO.setIdGeneratorModeName(idGeneratorMode.getDesc());
});
}
pageResult.setData(responseVOList);
@ -204,37 +208,37 @@ public class GroupConfigServiceImpl implements GroupConfigService {
return pageResult;
}
private boolean doSaveGroupConfig(UserSessionVO systemUser, GroupConfigRequestVO groupConfigRequestVO) {
private boolean doSaveGroupConfig(final String namespaceId, GroupConfigRequestVO groupConfigRequestVO) {
List<Integer> tablePartitionList = getTablePartitionList();
if (CollectionUtils.isEmpty(tablePartitionList)) {
return Boolean.FALSE;
}
GroupConfig groupConfig = GroupConfigConverter.INSTANCE.convert(groupConfigRequestVO);
GroupConfig groupConfig = GroupConfigConverter.INSTANCE.toGroupConfig(groupConfigRequestVO);
groupConfig.setCreateDt(LocalDateTime.now());
groupConfig.setVersion(1);
groupConfig.setNamespaceId(systemUser.getNamespaceId());
groupConfig.setNamespaceId(namespaceId);
groupConfig.setGroupName(groupConfigRequestVO.getGroupName());
groupConfig.setToken(groupConfigRequestVO.getToken());
groupConfig.setDescription(Optional.ofNullable(groupConfigRequestVO.getDescription()).orElse(StrUtil.EMPTY));
if (Objects.isNull(groupConfigRequestVO.getGroupPartition())) {
groupConfig.setGroupPartition(
HashUtil.bkdrHash(groupConfigRequestVO.getGroupName()) % tablePartitionList.size());
HashUtil.bkdrHash(groupConfigRequestVO.getGroupName()) % tablePartitionList.size());
} else {
Assert.isTrue(tablePartitionList.contains(groupConfigRequestVO.getGroupPartition()),
() -> new SnailJobServerException("分区不存在. [{}]", tablePartitionList));
() -> new SnailJobServerException("分区不存在. [{}]", tablePartitionList));
Assert.isTrue(groupConfigRequestVO.getGroupPartition() >= 0,
() -> new SnailJobServerException("分区不能是负数."));
() -> new SnailJobServerException("分区不能是负数."));
}
groupConfig.setBucketIndex(
HashUtil.bkdrHash(groupConfigRequestVO.getGroupName()) % systemProperties.getBucketTotal());
HashUtil.bkdrHash(groupConfigRequestVO.getGroupName()) % systemProperties.getBucketTotal());
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
Assert.isTrue(1 == groupConfigAccess.insert(groupConfig),
() -> new SnailJobServerException("新增组异常异常 groupConfigVO[{}]", groupConfigRequestVO));
() -> new SnailJobServerException("新增组异常异常 groupConfigVO[{}]", groupConfigRequestVO));
// 校验retry_task_x和retry_dead_letter_x是否存在
checkGroupPartition(groupConfig, systemUser.getNamespaceId());
checkGroupPartition(groupConfig, namespaceId);
return Boolean.TRUE;
}
@ -246,12 +250,12 @@ public class GroupConfigServiceImpl implements GroupConfigService {
try {
TaskAccess<RetryTask> retryTaskAccess = accessTemplate.getRetryTaskAccess();
retryTaskAccess.count(groupConfig.getGroupName(), namespaceId,
new LambdaQueryWrapper<RetryTask>().eq(RetryTask::getId, 1));
new LambdaQueryWrapper<RetryTask>().eq(RetryTask::getId, 1));
} catch (BadSqlGrammarException e) {
Optional.ofNullable(e.getMessage()).ifPresent(s -> {
if (s.contains("retry_task_" + groupConfig.getGroupPartition()) && s.contains("doesn't exist")) {
throw new SnailJobServerException("分区:[{}] '未配置表retry_task_{}', 请联系管理员进行配置",
groupConfig.getGroupPartition(), groupConfig.getGroupPartition());
groupConfig.getGroupPartition(), groupConfig.getGroupPartition());
}
});
}
@ -259,12 +263,12 @@ public class GroupConfigServiceImpl implements GroupConfigService {
try {
TaskAccess<RetryDeadLetter> retryTaskAccess = accessTemplate.getRetryDeadLetterAccess();
retryTaskAccess.one(groupConfig.getGroupName(), namespaceId,
new LambdaQueryWrapper<RetryDeadLetter>().eq(RetryDeadLetter::getId, 1));
new LambdaQueryWrapper<RetryDeadLetter>().eq(RetryDeadLetter::getId, 1));
} catch (BadSqlGrammarException e) {
Optional.ofNullable(e.getMessage()).ifPresent(s -> {
if (s.contains("retry_dead_letter_" + groupConfig.getGroupPartition()) && s.contains("doesn't exist")) {
throw new SnailJobServerException("分区:[{}] '未配置表retry_dead_letter_{}', 请联系管理员进行配置",
groupConfig.getGroupPartition(), groupConfig.getGroupPartition());
groupConfig.getGroupPartition(), groupConfig.getGroupPartition());
}
});
}
@ -275,12 +279,12 @@ public class GroupConfigServiceImpl implements GroupConfigService {
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
GroupConfig groupConfig = groupConfigAccess.one(
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(GroupConfig::getGroupName, groupName));
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(GroupConfig::getGroupName, groupName));
GroupConfigResponseVO groupConfigResponseVO = GroupConfigResponseVOConverter.INSTANCE.convert(
groupConfig);
groupConfig);
Optional.ofNullable(IdGeneratorModeEnum.modeOf(groupConfig.getIdGeneratorMode())).ifPresent(idGeneratorMode -> {
groupConfigResponseVO.setIdGeneratorModeName(idGeneratorMode.getDesc());
@ -295,21 +299,21 @@ public class GroupConfigServiceImpl implements GroupConfigService {
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
List<GroupConfig> groupConfigs = groupConfigAccess.list(
new LambdaQueryWrapper<GroupConfig>()
.select(GroupConfig::getGroupName, GroupConfig::getNamespaceId)
.in(CollUtil.isNotEmpty(namespaceIds), GroupConfig::getNamespaceId, namespaceIds));
new LambdaQueryWrapper<GroupConfig>()
.select(GroupConfig::getGroupName, GroupConfig::getNamespaceId)
.in(CollUtil.isNotEmpty(namespaceIds), GroupConfig::getNamespaceId, namespaceIds));
if (CollUtil.isEmpty(groupConfigs)) {
return Collections.emptyList();
}
List<Namespace> namespaces = namespaceMapper.selectList(
new LambdaQueryWrapper<Namespace>()
.in(Namespace::getUniqueId, StreamUtils.toSet(groupConfigs, GroupConfig::getNamespaceId)));
new LambdaQueryWrapper<Namespace>()
.in(Namespace::getUniqueId, StreamUtils.toSet(groupConfigs, GroupConfig::getNamespaceId)));
Map<String, String> namespaceMap = StreamUtils.toMap(namespaces, Namespace::getUniqueId, Namespace::getName);
List<GroupConfigResponseVO> groupConfigResponses = GroupConfigResponseVOConverter.INSTANCE.convertList(
groupConfigs);
groupConfigs);
for (final GroupConfigResponseVO groupConfigResponseVO : groupConfigResponses) {
groupConfigResponseVO.setNamespaceName(namespaceMap.get(groupConfigResponseVO.getNamespaceId()));
}
@ -327,8 +331,8 @@ public class GroupConfigServiceImpl implements GroupConfigService {
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
List<GroupConfig> groupConfigs = groupConfigAccess.list(new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, userSessionVO.getNamespaceId())
.select(GroupConfig::getGroupName));
.eq(GroupConfig::getNamespaceId, userSessionVO.getNamespaceId())
.select(GroupConfig::getGroupName));
return StreamUtils.toList(groupConfigs, GroupConfig::getGroupName);
}
@ -336,9 +340,9 @@ public class GroupConfigServiceImpl implements GroupConfigService {
@Override
public List<String> getOnlinePods(String groupName) {
List<ServerNode> serverNodes = serverNodeMapper.selectList(
new LambdaQueryWrapper<ServerNode>()
.eq(ServerNode::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(ServerNode::getGroupName, groupName));
new LambdaQueryWrapper<ServerNode>()
.eq(ServerNode::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.eq(ServerNode::getGroupName, groupName));
return StreamUtils.toList(serverNodes, serverNode -> serverNode.getHostIp() + ":" + serverNode.getHostPort());
}
@ -368,8 +372,8 @@ public class GroupConfigServiceImpl implements GroupConfigService {
}
return tableList.stream().map(ReUtil::getFirstNumber).filter(i ->
!Objects.isNull(i)).distinct()
.collect(Collectors.toList());
!Objects.isNull(i)).distinct()
.collect(Collectors.toList());
} catch (SQLException ignored) {
} finally {
if (Objects.nonNull(connection)) {
@ -383,4 +387,45 @@ public class GroupConfigServiceImpl implements GroupConfigService {
return Lists.newArrayList();
}
@Override
@Transactional
public void importGroup(final List<GroupConfigRequestVO> requestList) {
String namespaceId = UserSessionUtils.currentUserSession().getNamespaceId();
Set<String> groupSet = StreamUtils.toSet(requestList, GroupConfigRequestVO::getGroupName);
ConfigAccess<GroupConfig> groupConfigAccess = accessTemplate.getGroupConfigAccess();
Assert.isTrue(groupConfigAccess.count(new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.in(GroupConfig::getGroupName, groupSet)) == 0,
() -> new SnailJobServerException("GroupName已经存在 {}", groupSet));
for (final GroupConfigRequestVO groupConfigRequestVO : requestList) {
// 保存组配置
doSaveGroupConfig(namespaceId, groupConfigRequestVO);
// 保存生成唯一id配置
doSaveSequenceAlloc(namespaceId, groupConfigRequestVO);
}
}
@Override
public String exportGroup(final Set<Long> groupIds) {
String namespaceId = UserSessionUtils.currentUserSession().getNamespaceId();
List<GroupConfig> groupConfigs = accessTemplate.getGroupConfigAccess().list(
new LambdaQueryWrapper<GroupConfig>()
.eq(GroupConfig::getNamespaceId, namespaceId)
.in(GroupConfig::getId, groupIds)
);
SetView<Long> notExistedGroupIdSet = Sets.difference(groupIds,
StreamUtils.toSet(groupConfigs, GroupConfig::getId));
Assert.isTrue(groupIds.size() == groupConfigs.size(), () -> new SnailJobServerException("导出失败. 组ID{}不存在", notExistedGroupIdSet));
return JsonUtil.toJsonString(GroupConfigConverter.INSTANCE.toGroupConfigRequestVOs(groupConfigs));
}
}

42
snail-job-server/snail-job-server-web/src/main/java/com/aizuda/snailjob/server/web/service/impl/SceneConfigServiceImpl.java
View File

@ -22,7 +22,6 @@ import com.aizuda.snailjob.template.datasource.access.AccessTemplate;
import com.aizuda.snailjob.template.datasource.access.ConfigAccess;
import com.aizuda.snailjob.template.datasource.persistence.mapper.NamespaceMapper;
import com.aizuda.snailjob.template.datasource.persistence.po.GroupConfig;
import com.aizuda.snailjob.template.datasource.persistence.po.Namespace;
import com.aizuda.snailjob.template.datasource.persistence.po.RetrySceneConfig;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
@ -31,6 +30,7 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.collect.Sets.SetView;
import lombok.RequiredArgsConstructor;
import org.checkerframework.checker.units.qual.C;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@ -190,35 +190,17 @@ public class SceneConfigServiceImpl implements SceneConfigService {
List<RetrySceneConfig> sceneConfigs = accessTemplate.getSceneConfigAccess()
.list(new LambdaQueryWrapper<RetrySceneConfig>()
.eq(RetrySceneConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.in(RetrySceneConfig::getId, sceneIds)
// TODO 若导出全部需要分页查询，避免一次拉取太多数据
.in(CollUtil.isNotEmpty(sceneIds), RetrySceneConfig::getId, sceneIds)
);
List<SceneConfigRequestVO> sceneConfigRequestVOs = SceneConfigConverter.INSTANCE.toSceneConfigRequestVOs(
sceneConfigs);
return JsonUtil.toJsonString(sceneConfigRequestVOs);
SetView<Long> notExistedSceneIdSet = Sets.difference(sceneIds,
StreamUtils.toSet(sceneConfigs, RetrySceneConfig::getId));
Assert.isTrue(sceneIds.size() == sceneConfigs.size(), () -> new SnailJobServerException("导出失败. 场景ID{}不存在", notExistedSceneIdSet));
return JsonUtil.toJsonString(SceneConfigConverter.INSTANCE.toSceneConfigRequestVOs(sceneConfigs));
}
@Override
@Transactional
public void batchCopy(final Long targetNamespaceId, final Set<Long> sceneIds) {
Assert.notEmpty(sceneIds, () -> new SnailJobServerException("参数错误"));
Namespace namespace = namespaceMapper.selectById(targetNamespaceId);
Assert.notNull(namespace, () -> new SnailJobServerException("空间不存在"));
List<RetrySceneConfig> sceneConfigs = accessTemplate.getSceneConfigAccess()
.list(new LambdaQueryWrapper<RetrySceneConfig>()
.eq(RetrySceneConfig::getNamespaceId, UserSessionUtils.currentUserSession().getNamespaceId())
.in(RetrySceneConfig::getId, sceneIds)
);
Assert.isTrue(sceneIds.size() == sceneConfigs.size(), () -> new SnailJobServerException("存在部分场景配置不存在"));
List<SceneConfigRequestVO> sceneConfigRequestVOs = SceneConfigConverter.INSTANCE.toSceneConfigRequestVOs(
sceneConfigs);
// 批量写入
batchSaveSceneConfig(sceneConfigRequestVOs, namespace.getUniqueId());
}
private void batchSaveSceneConfig(final List<SceneConfigRequestVO> requests, final String namespaceId) {
@ -240,8 +222,8 @@ public class SceneConfigServiceImpl implements SceneConfigService {
SetView<String> notExistedGroupNameSet = Sets.difference(groupNameSet,
StreamUtils.toSet(groupConfigs, GroupConfig::getGroupName));
Assert.notEmpty(notExistedGroupNameSet,
() -> new SnailJobServerException("组:{}不存在", notExistedGroupNameSet));
Assert.isTrue(CollUtil.isEmpty(notExistedGroupNameSet),
() -> new SnailJobServerException("导入失败. 原因: 组{}不存在", notExistedGroupNameSet));
ConfigAccess<RetrySceneConfig> sceneConfigAccess = accessTemplate.getSceneConfigAccess();
List<RetrySceneConfig> sceneConfigs = sceneConfigAccess.list(
@ -249,9 +231,9 @@ public class SceneConfigServiceImpl implements SceneConfigService {
.select(RetrySceneConfig::getSceneName)
.eq(RetrySceneConfig::getNamespaceId, namespaceId)
.in(RetrySceneConfig::getGroupName, groupNameSet)
.eq(RetrySceneConfig::getSceneName, sceneNameSet));
.in(RetrySceneConfig::getSceneName, sceneNameSet));
Assert.isTrue(CollUtil.isEmpty(sceneConfigs), () -> new SnailJobServerException("场景:{}已存在",
Assert.isTrue(CollUtil.isEmpty(sceneConfigs), () -> new SnailJobServerException("导入失败. 原因:场景{}已存在",
StreamUtils.toSet(sceneConfigs, RetrySceneConfig::getSceneName)));
LocalDateTime now = LocalDateTime.now();