Gitlink
/
gitlink-notification-system
0
0
Fork 4
Code Issues Pull Requests Releases 1 Wiki Activity

Compare commits

base: Gitlink:master
Branches Tags
Gitlink:master
Gitlink:dev-nacos
Gitlink:dev-k8s
Gitlink:release
DavidZeng:dev_add_middleware_conf
DavidZeng:master
Gitlink:v2109290953
..
compare: DavidZeng:master
Branches Tags
DavidZeng:dev_add_middleware_conf
DavidZeng:master
Gitlink:master
Gitlink:dev-nacos
Gitlink:dev-k8s
Gitlink:release
Gitlink:v2109290953

No commits in common. "master" and "master" have entirely different histories.
master ... master

33 changed files with 383 additions and 440 deletions
Show Stats Expand all files Collapse all files

3
README.md
View File

@ -33,9 +33,6 @@ cp middleware/.env.example middleware/.env
cp reader/src/main/resources/application.yml.example reader/src/main/resources/application.yml cp reader/src/main/resources/application.yml.example reader/src/main/resources/application.yml
cp writer/src/main/resources/application.yml.example writer/src/main/resources/application.yml cp writer/src/main/resources/application.yml.example writer/src/main/resources/application.yml
cp executor/src/main/resources/application.yml.example executor/src/main/resources/application.yml cp executor/src/main/resources/application.yml.example executor/src/main/resources/application.yml
cp reader/src/main/resources/mail.properties.example reader/src/main/resources/mail.properties
cp writer/src/main/resources/mail.properties.example writer/src/main/resources/mail.properties
cp executor/src/main/resources/mail.properties.example executor/src/main/resources/mail.properties
``` ```
5. 修改 `{repo}/middleware/.env` 文件里 `SQL_SCRIPT_PATH` 和 `DOCKER_DATA_PATH` 绝对路径到本地磁盘 5. 修改 `{repo}/middleware/.env` 文件里 `SQL_SCRIPT_PATH` 和 `DOCKER_DATA_PATH` 绝对路径到本地磁盘

17
common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java
View File

@ -33,17 +33,12 @@ public class KafkaProducerConfig {
@Bean @Bean
ProducerFactory<String, String> producerConfigs() { ProducerFactory<String, String> producerConfigs() {
Map<String, Object> props = new HashMap<>(); Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
// 如果当前配置信息内没有kafka producer相关配置则不做参数设置，此处解决没有设置配置信息而引起的空指针异常 props.put(ProducerConfig.RETRIES_CONFIG, retries);
if (null != bootstrapServers) { props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.RETRIES_CONFIG, retries); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
}
return new DefaultKafkaProducerFactory<>(props); return new DefaultKafkaProducerFactory<>(props);
} }
} }

25
common/src/main/java/cn/org/gitlink/notification/common/constant/NotificationSystemConstant.java
View File

@ -4,8 +4,7 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
/** /**
* 消息系统常量 * @description:
*
* @author: wanjia * @author: wanjia
* @time: 2021/9/9 * @time: 2021/9/9
*/ */
@ -13,32 +12,10 @@ public class NotificationSystemConstant {
//平台编码 //平台编码
public static final String PLATFORM_CODE_GITLINK = "gitlink"; //gitlink平台 public static final String PLATFORM_CODE_GITLINK = "gitlink"; //gitlink平台
public static final String PLATFORM_CODE_HEHUI = "hehui"; //hehui平台 public static final String PLATFORM_CODE_HEHUI = "hehui"; //hehui平台
public static final String PLATFORM_CODE_OSREDM = "osredm"; //红山开源平台
public static final Map<String, String> PLATFORM_CODE_MAP = new HashMap<String, String>() { public static final Map<String, String> PLATFORM_CODE_MAP = new HashMap<String, String>() {
{ {
put("gitlink", PLATFORM_CODE_GITLINK); put("gitlink", PLATFORM_CODE_GITLINK);
put("hehui", PLATFORM_CODE_HEHUI); put("hehui", PLATFORM_CODE_HEHUI);
put("osredm", PLATFORM_CODE_OSREDM);
} }
}; };
//邮件任务处理状态
public static final Integer EMAIL_JOB_NOT_DISPATCHED = -1; //处理成功
public static final Integer EMAIL_JOB_DISPATCHED_SUCCESS = 1; //处理成功
public static final Integer EMAIL_JOB_DISPATCHED_FAIL = 2; //处理失败
//邮件发送记录状态
public static final Integer EMAIL_UNSENT_RECORD = -1; //处理成功
public static final Integer EMAIL_SENT_SUCCESS = 1; //处理成功
public static final Integer EMAIL_SENT_FAIL = 2; //处理失败
//未读消息类型
public static final Integer NOTIFICATION_TYPE_ALL = -1; //全部未读消息
public static final Integer NOTIFICATION_TYPE_SYS = 1; //系统消息
public static final Integer NOTIFICATION_TYPE_ATME = 2; //@我的消息
//是否已读状态
public static final Integer STATUS_OPTION_ALL = -1; //所有信息类型，包括未读、已读
public static final Integer STATUS_OPTION_UNREAD = 1; //未读信息
public static final Integer STATUS_OPTION_READ = 2; //已读信息
} }

25
common/src/main/java/cn/org/gitlink/notification/common/interceptor/ExceptionInterceptor.java
View File

@ -7,16 +7,11 @@ import cn.org.gitlink.notification.common.utils.CommonUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.ResponseStatus;
import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/** /**
* web请求异常全局处理类，在使用的模块内继承此类，然后加上ControllerAdvice注解，example: * web请求异常全局处理类，在使用的模块内继承此类，然后加上ControllerAdvice注解，example:
@ -44,24 +39,4 @@ public class ExceptionInterceptor {
logger.error(logTemplate, request.getRequestURI(), e); logger.error(logTemplate, request.getRequestURI(), e);
return DataPacketUtil.jsonFailResult(e.getMessage()); return DataPacketUtil.jsonFailResult(e.getMessage());
} }
/**
* valid 异常处理
*
* @param e
* @return
*/
@ResponseBody
@ResponseStatus(HttpStatus.OK)
@ExceptionHandler(MethodArgumentNotValidException.class)
public ResponseData handlerValidException(MethodArgumentNotValidException e){
Map<String, Object> errorMessageMap = new HashMap<>();
// 获取所有错误信息
List<FieldError> errList = e.getBindingResult().getFieldErrors();
for (FieldError error : errList) {
errorMessageMap.put(error.getField(), error.getDefaultMessage());
}
logger.error("data errors:{}", errorMessageMap);
return DataPacketUtil.jsonFailResult(errorMessageMap);
}
} }

1
common/src/main/java/cn/org/gitlink/notification/common/response/DataPacketUtil.java
View File

@ -2,6 +2,7 @@ package cn.org.gitlink.notification.common.response;
import java.util.HashMap; import java.util.HashMap;
import java.util.List;
public class DataPacketUtil public class DataPacketUtil
{ {

34
common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java
View File

@ -14,7 +14,10 @@ import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.util.concurrent.ListenableFutureCallback;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import java.util.*; import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -26,7 +29,7 @@ import java.util.stream.Collectors;
* @date 2021-09-07 * @date 2021-09-07
*/ */
@Component(value = "GitlinkKafkaUtil") @Component
public class KafkaUtil { public class KafkaUtil {
private Logger logger = LogManager.getLogger(KafkaUtil.class); private Logger logger = LogManager.getLogger(KafkaUtil.class);
@ -34,15 +37,6 @@ public class KafkaUtil {
@Value("${spring.kafka.producer.bootstrap_servers:#{null}}") @Value("${spring.kafka.producer.bootstrap_servers:#{null}}")
private String kafkaServer; private String kafkaServer;
@Value("${spring.kafka.producer.partitions:#{null}}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor:#{null}}")
private Short replicationFactor;
@Value("${spring.kafka.producer.topics:''}")
private String topicString;
private AdminClient adminClient; private AdminClient adminClient;
@Autowired @Autowired
@ -52,32 +46,16 @@ public class KafkaUtil {
* 初始化AdminClient * 初始化AdminClient
* '@PostConstruct该注解被用来修饰一个非静态的void（）方法。 * '@PostConstruct该注解被用来修饰一个非静态的void（）方法。
* 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。 * 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。
* PostConstruct在构造函数之后执行，init（）方法之前执行。ls * PostConstruct在构造函数之后执行，init（）方法之前执行。
*/ */
@PostConstruct @PostConstruct
private void initAdminClient() { private void initAdminClient() {
Map<String, Object> props = new HashMap<>(1); Map<String, Object> props = new HashMap<>(1);
// 如果当前配置信息内没有kafka producer相关配置则不对adminClient做初始化
if (kafkaServer == null) { if (kafkaServer == null) {
return; return;
} }
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
adminClient = KafkaAdminClient.create(props); adminClient = KafkaAdminClient.create(props);
// 初始化topics
if (null !=partitions && null != replicationFactor) {
List<String> topics = Arrays.asList(topicString.split(","));
if (!topics.isEmpty()) {
List<NewTopic> newTopics = new LinkedList<>();
topics.forEach(topic -> {
newTopics.add(new NewTopic(topic, partitions, replicationFactor));
});
this.createTopic(newTopics);
}
}
} }
public void createTopic(Collection<NewTopic> newTopics) { public void createTopic(Collection<NewTopic> newTopics) {

19
common/src/main/java/cn/org/gitlink/notification/common/utils/ValidatorUtils.java
View File

@ -1,9 +1,5 @@
package cn.org.gitlink.notification.common.utils; package cn.org.gitlink.notification.common.utils;
import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
import cn.org.gitlink.notification.common.response.DataPacketUtil;
import cn.org.gitlink.notification.common.response.ResponseData;
import org.springframework.validation.BindingResult; import org.springframework.validation.BindingResult;
import org.springframework.validation.FieldError; import org.springframework.validation.FieldError;
@ -25,19 +21,4 @@ public class ValidatorUtils {
} }
return errorMessageMap; return errorMessageMap;
} }
/**
* 验证 platform合法性
*
* @param platform
* @return
*/
public static ResponseData validatePlatformCode(String platform) {
//验证 {platform} 参数合法性，以判断请求来源
if (!NotificationSystemConstant.PLATFORM_CODE_MAP.containsKey(platform)) {
return DataPacketUtil.jsonFailResult("{platform} 参数非法");
}
return null;
}
} }

7
db/gns-email.sql
View File

@ -23,9 +23,4 @@ CREATE TABLE `gitlink_email_send_records` (
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
KEY `index_on_email_and_status` (`email`,`status`), KEY `index_on_email_and_status` (`email`,`status`),
KEY `index_on_status` (`status`) KEY `index_on_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3; ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;
-- 2023-01-05 更新字符集编码
ALTER TABLE gitlink_email_jobs CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
ALTER TABLE gitlink_email_send_records CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

5
db/gns-notification.sql
View File

@ -35,7 +35,4 @@ ALTER TABLE gitlink_sys_notification ADD COLUMN (`type` TINYINT(4) NOT NULL DEFA
-- 2021-09-09 新增 source 字段区分消息来源、新增 extra 字段保存额外信息 -- 2021-09-09 新增 source 字段区分消息来源、新增 extra 字段保存额外信息
ALTER TABLE gitlink_sys_notification ADD source varchar(250) NULL COMMENT '消息来源'; ALTER TABLE gitlink_sys_notification ADD source varchar(250) NULL COMMENT '消息来源';
ALTER TABLE gitlink_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）'; ALTER TABLE gitlink_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）';
-- 2023-01-05 更新字符集编码
ALTER TABLE gitlink_sys_notification CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

5
db/hehui-gns-notification.sql
View File

@ -24,7 +24,4 @@ ALTER TABLE hehui_sys_notification ADD COLUMN (`type` TINYINT(4) NOT NULL DEFAUL
-- 2021-09-10 新增 source 字段区分消息来源、新增 extra 字段保存额外信息 -- 2021-09-10 新增 source 字段区分消息来源、新增 extra 字段保存额外信息
ALTER TABLE hehui_sys_notification ADD source varchar(250) NULL COMMENT '消息来源'; ALTER TABLE hehui_sys_notification ADD source varchar(250) NULL COMMENT '消息来源';
ALTER TABLE hehui_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）'; ALTER TABLE hehui_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）';
-- 2023-01-05 更新字符集编码
ALTER TABLE hehui_sys_notification CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

30
db/osredm-gns-notification.sql
View File

@ -1,30 +0,0 @@
USE gitlink_notification;
-- 2022-02-14 新增红山平台
INSERT INTO gns_platform_info(platform_code,platform_name,created_at,is_delete) VALUES('osredm','红山平台',NOW(),-1);
DROP TABLE IF EXISTS `osredm_sys_notification`;
CREATE TABLE `osredm_sys_notification` (
`id` INT NOT NULL AUTO_INCREMENT,
`sender` INT(11) NOT NULL COMMENT '发送者id',
`receiver` INT(11) NOT NULL COMMENT '接受者id',
`content` TEXT NOT NULL COMMENT '消息内容：富文本',
`notification_url` VARCHAR(2000) DEFAULT NULL COMMENT '消息跳转链接',
`created_at` DATETIME NOT NULL DEFAULT NOW() COMMENT '创建时间',
`status` TINYINT(4) NOT NULL DEFAULT 1 COMMENT '已读状态: 1未读，2已读',
`is_delete` TINYINT(1) NOT NULL DEFAULT '-1' COMMENT '是否删除: -1未删除，1已删除',
PRIMARY KEY (`id`),
KEY `index_on_receiver_and_status` (`receiver`,`status`),
KEY `index_on_status` (`status`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb3;
-- 2021-09-10 区分系统消息类型
ALTER TABLE osredm_sys_notification ADD COLUMN (`type` TINYINT(4) NOT NULL DEFAULT 1 COMMENT '消息类型： 1系统消息，2@我');
-- 2021-09-10 新增 source 字段区分消息来源、新增 extra 字段保存额外信息
ALTER TABLE osredm_sys_notification ADD source varchar(250) NULL COMMENT '消息来源';
ALTER TABLE osredm_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）';
-- 2023-01-05 更新字符集编码
ALTER TABLE osredm_sys_notification CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

28
common/src/main/java/cn/org/gitlink/notification/common/config/KafkaConsumerConfig.java → executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java
View File

@ -1,4 +1,4 @@
package cn.org.gitlink.notification.common.config; package cn.org.gitlink.notification.executor.core.config;
import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringDeserializer;
@ -20,21 +20,17 @@ public class KafkaConsumerConfig {
@Value("${spring.kafka.consumer.bootstrap_servers:#{null}}") @Value("${spring.kafka.consumer.bootstrap_servers:#{null}}")
private String servers; private String servers;
@Value("${spring.kafka.consumer.auto_offset_reset:#{null}}") @Value("${spring.kafka.consumer.auto_offset_reset}")
private String autoOffsetReset; private String autoOffsetReset;
@Value("${spring.kafka.consumer.max_poll_records:#{null}") @Value("${spring.kafka.consumer.max_poll_records}")
private String maxPollRecords; private String maxPollRecords;
@Value("${spring.kafka.consumer.topic:#{null}}") @Value("${spring.kafka.consumer.topic}")
private String topic; private String topic;
@Bean @Bean
public ConcurrentKafkaListenerContainerFactory<String, String> consumerListenerFactory() { public ConcurrentKafkaListenerContainerFactory<String, String> consumerListenerFactory() {
// 如果当前配置信息内没有kafka consumer则不创建监听工厂
if (null == servers) return null;
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerConfigs()); factory.setConsumerFactory(consumerConfigs());
factory.setRecordFilterStrategy(record -> record.topic().toLowerCase().equals(this.topic)); factory.setRecordFilterStrategy(record -> record.topic().toLowerCase().equals(this.topic));
@ -44,16 +40,12 @@ public class KafkaConsumerConfig {
@Bean @Bean
public ConsumerFactory<String, Object> consumerConfigs() { public ConsumerFactory<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>(); Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
// 如果当前配置信息内没有kafka consumer，此处解决没有设置配置信息而引起的空指针异常 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
if (null != servers) { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
}
return new DefaultKafkaConsumerFactory<>(props); return new DefaultKafkaConsumerFactory<>(props);
} }
} }

24
executor/src/main/java/cn/org/gitlink/notification/executor/service/email/EmailService.java
View File

@ -1,6 +1,5 @@
package cn.org.gitlink.notification.executor.service.email; package cn.org.gitlink.notification.executor.service.email;
import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
import cn.org.gitlink.notification.common.utils.EmailUtils; import cn.org.gitlink.notification.common.utils.EmailUtils;
import cn.org.gitlink.notification.model.dao.entity.EmailJob; import cn.org.gitlink.notification.model.dao.entity.EmailJob;
import cn.org.gitlink.notification.model.dao.entity.EmailSendRecord; import cn.org.gitlink.notification.model.dao.entity.EmailSendRecord;
@ -22,6 +21,16 @@ public class EmailService {
private Logger logger = LogManager.getLogger(EmailService.class); private Logger logger = LogManager.getLogger(EmailService.class);
//邮件任务处理状态
private static final Integer EMAIL_JOB_NOT_DISPATCHED = -1; //处理成功
private static final Integer EMAIL_JOB_DISPATCHED_SUCCESS = 1; //处理成功
private static final Integer EMAIL_JOB_DISPATCHED_FAIL = 2; //处理失败
//邮件发送记录状态
private static final Integer EMAIL_UNSENT_RECORD = -1; //处理成功
private static final Integer EMAIL_SENT_SUCCESS = 1; //处理成功
private static final Integer EMAIL_SENT_FAIL = 2; //处理失败
@Autowired @Autowired
private EmailJobsService emailJobsService; private EmailJobsService emailJobsService;
@ -45,7 +54,7 @@ public class EmailService {
//获取指定数量待处理列表 //获取指定数量待处理列表
List<EmailJob> emailJobList = new ArrayList<>(); List<EmailJob> emailJobList = new ArrayList<>();
try { try {
emailJobList = emailJobsService.getEmailJobsByDispatchedStatus(platform, NotificationSystemConstant.EMAIL_JOB_NOT_DISPATCHED, dispatchNumber); emailJobList = emailJobsService.getEmailJobsByDispatchedStatus(platform, EMAIL_JOB_NOT_DISPATCHED, dispatchNumber);
} catch (Exception e) { } catch (Exception e) {
logger.error("获取未处理邮件任务列表失败：\n" + e); logger.error("获取未处理邮件任务列表失败：\n" + e);
} }
@ -54,13 +63,13 @@ public class EmailService {
Boolean flag = null; Boolean flag = null;
for (EmailJob emailJob : emailJobList) { for (EmailJob emailJob : emailJobList) {
try { try {
flag = emailSendRecordsService.createEmailSendRecords(platform, emailJob.getEmails(), emailJob.getId()); flag = emailSendRecordsService.newEmailSendRecords(platform, emailJob.getEmails(), emailJob.getId());
} catch (Exception e) { } catch (Exception e) {
logger.error("处理EmailJob失败，email_job_id: " + emailJob.getId() + "\n" + e); logger.error("处理EmailJob失败，email_job_id: " + emailJob.getId() + "\n" + e);
} }
//EmailJob分配成功，更新状态 //EmailJob分配成功，更新状态
try { try {
emailJobsService.markEmailJobAs(platform, emailJob.getId(), new Date(), flag ? NotificationSystemConstant.EMAIL_JOB_DISPATCHED_SUCCESS : NotificationSystemConstant.EMAIL_JOB_DISPATCHED_FAIL); emailJobsService.markEmailJobsAs(platform, emailJob.getId(), new Date(), flag ? EMAIL_JOB_DISPATCHED_SUCCESS : EMAIL_JOB_DISPATCHED_FAIL);
} catch (Exception e) { } catch (Exception e) {
logger.error("更新EmailJob状态失败，email_job_id: " + emailJob.getId() + "\n" + e); logger.error("更新EmailJob状态失败，email_job_id: " + emailJob.getId() + "\n" + e);
} }
@ -81,7 +90,7 @@ public class EmailService {
//获取待发送列表 //获取待发送列表
List<EmailSendRecord> emailSendRecordList = new ArrayList<>(); List<EmailSendRecord> emailSendRecordList = new ArrayList<>();
try { try {
emailSendRecordList = emailSendRecordsService.getRecordsByStatus(platform, NotificationSystemConstant.EMAIL_UNSENT_RECORD, sentNumber); emailSendRecordList = emailSendRecordsService.getRecordsByStatus(platform, EMAIL_UNSENT_RECORD, sentNumber);
} catch (Exception e) { } catch (Exception e) {
logger.error("获取未发送邮件列表失败：\n" + e); logger.error("获取未发送邮件列表失败：\n" + e);
} }
@ -94,9 +103,8 @@ public class EmailService {
emailUtils.sendMail(unSentEmailSendRecord.getSubject(), unSentEmailSendRecord.getEmail(), unSentEmailSendRecord.getContent()); emailUtils.sendMail(unSentEmailSendRecord.getSubject(), unSentEmailSendRecord.getEmail(), unSentEmailSendRecord.getContent());
flag = true; flag = true;
unSentEmailSendRecord.setSentAt(new Date()); unSentEmailSendRecord.setSentAt(new Date());
unSentEmailSendRecord.setStatus(flag ? NotificationSystemConstant.EMAIL_SENT_SUCCESS : NotificationSystemConstant.EMAIL_SENT_FAIL); unSentEmailSendRecord.setStatus(flag ? EMAIL_SENT_SUCCESS : EMAIL_SENT_FAIL);
} catch (Exception e) { } catch (MessagingException e) {
unSentEmailSendRecord.setStatus(NotificationSystemConstant.EMAIL_SENT_FAIL);
logger.error("发送邮件失败，email: " + unSentEmailSendRecord.getEmail() + "\n" + e); logger.error("发送邮件失败，email: " + unSentEmailSendRecord.getEmail() + "\n" + e);
} }
} }

14
executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java
View File

@ -4,16 +4,18 @@ import cn.org.gitlink.notification.common.utils.KafkaUtil;
import cn.org.gitlink.notification.model.dao.entity.vo.NewEmailJobVo; import cn.org.gitlink.notification.model.dao.entity.vo.NewEmailJobVo;
import cn.org.gitlink.notification.model.service.notification.EmailJobsService; import cn.org.gitlink.notification.model.service.notification.EmailJobsService;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaHandler; import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.Arrays;
@Component @Component
@Configuration @Configuration
@KafkaListener(topics = "${spring.kafka.consumer.topic_email}", groupId = "${spring.kafka.consumer.group_id_email}") @KafkaListener(topics = "${spring.kafka.consumer.topic_email}", groupId = "${spring.kafka.consumer.group_id_email}")
@ -27,17 +29,23 @@ public class EmailJobsListener {
@Value("${spring.kafka.producer.topic_new_email_remind}") @Value("${spring.kafka.producer.topic_new_email_remind}")
private String gitlinkNewEmailRemindTopic; private String gitlinkNewEmailRemindTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired @Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil; private KafkaUtil kafkaUtil;
@KafkaHandler @KafkaHandler
public void messageHandler(String message) { public void messageHandler(String message) {
try { try {
NewEmailJobVo newEmailJobVo = JSONObject.parseObject(message, NewEmailJobVo.class); NewEmailJobVo newEmailJobVo = JSONObject.parseObject(message, NewEmailJobVo.class);
Boolean flag = emailJobsService.createEmailJob(newEmailJobVo); Boolean flag = emailJobsService.sendEmail(newEmailJobVo);
//if the message is inserted successfully, send a new email-job message to kafka //if the message is inserted successfully, send a new email-job message to kafka
if (flag){ if (flag){
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNewEmailRemindTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkNewEmailRemindTopic, JSONObject.toJSONString(newEmailJobVo)); kafkaUtil.sendMessage(gitlinkNewEmailRemindTopic, JSONObject.toJSONString(newEmailJobVo));
} }
} catch (Exception e) { } catch (Exception e) {

4
executor/src/main/resources/application.yml.example
View File

@ -13,8 +13,6 @@ spring:
replication_factor: 1 replication_factor: 1
partitions: 3 partitions: 3
topic_new_email_remind: topic-gitlink-new-email-remind topic_new_email_remind: topic-gitlink-new-email-remind
# 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
topics: ${spring.kafka.producer.topic_new_email_remind}
consumer: consumer:
bootstrap_servers: kafka1:9092,kafka2:9092 bootstrap_servers: kafka1:9092,kafka2:9092
group_id: group-gitlink-notification group_id: group-gitlink-notification
@ -28,7 +26,7 @@ spring:
topic_email: topic-gitlink-email topic_email: topic-gitlink-email
topic_new_email_remind: topic-gitlink-new-email-remind topic_new_email_remind: topic-gitlink-new-email-remind
listener: listener:
concurrency: 1 concurrency: 3
ack-mode: record ack-mode: record
redis: redis:

0
middleware/end_docker_compose.sh Normal file → Executable file
View File

353
middleware/services.yml
View File

@ -1,181 +1,172 @@
version: '3' version: '3'
services: services:
mysql: mysql:
image: mysql:${MYSQL_VERSION} image: mysql:${MYSQL_VERSION}
container_name: ${MYSQL_CONTAINER_NAME} container_name: ${MYSQL_CONTAINER_NAME}
hostname: mysql hostname: mysql
environment: environment:
- MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD} - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
- MYSQL_USER=${MYSQL_USER} - MYSQL_USER=${MYSQL_USER}
- MYSQL_PASSWORD=${MYSQL_PASSWORD} - MYSQL_PASSWORD=${MYSQL_PASSWORD}
- MYSQL_DATABASE=${MYSQL_DATABASE} - MYSQL_DATABASE=${MYSQL_DATABASE}
- TZ=Asia/Shanghai - TZ=Asia/Shanghai
volumes: volumes:
- ${DOCKER_DATA_PATH}/mysql:/var/lib/mysql - ${DOCKER_DATA_PATH}/mysql:/var/lib/mysql
- ${SQL_SCRIPT_PATH}/gns-notification.sql:/docker-entrypoint-initdb.d/0001.sql - ${SQL_SCRIPT_PATH}/gns-notification.sql:/docker-entrypoint-initdb.d/0001.sql
- ${SQL_SCRIPT_PATH}/hehui-gns-notification.sql:/docker-entrypoint-initdb.d/0002.sql - ${SQL_SCRIPT_PATH}/hehui-gns-notification.sql:/docker-entrypoint-initdb.d/0002.sql
- ${SQL_SCRIPT_PATH}/gns-email.sql:/docker-entrypoint-initdb.d/0003.sql - ${SQL_SCRIPT_PATH}/gns-email.sql:/docker-entrypoint-initdb.d/0003.sql
- ${SQL_SCRIPT_PATH}/osredm-gns-notification.sql:/docker-entrypoint-initdb.d/0004.sql command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci ports:
ports: - ${MYSQL_LOCAL_PORT}:3306
- ${MYSQL_LOCAL_PORT}:3306 networks:
networks: - gitlink_network
- gitlink_network
restart: always redis:
image: redis:${REDIS_VERSION}
redis: container_name: ${REDIS_CONTAINER_NAME}
image: redis:${REDIS_VERSION} hostname: redis
container_name: ${REDIS_CONTAINER_NAME} volumes:
hostname: redis - ${DOCKER_DATA_PATH}/redis/data:/data
volumes: - ${DOCKER_DATA_PATH}/redis/logs:/logs
- ${DOCKER_DATA_PATH}/redis/data:/data environment:
- ${DOCKER_DATA_PATH}/redis/logs:/logs - TZ=Asia/Shanghai
environment: ports:
- TZ=Asia/Shanghai - ${REDIS_LOCAL_PORT}:6379
ports: networks:
- ${REDIS_LOCAL_PORT}:6379 - gitlink_network
networks:
- gitlink_network # See Also: https://hub.docker.com/r/confluentinc/cp-zookeeper
restart: always zookeeper:
image: confluentinc/cp-zookeeper:latest
# See Also: https://hub.docker.com/r/confluentinc/cp-zookeeper container_name: ${ZOOKEEPER_CONTAINER_NAME}
zookeeper: hostname: zookeeper
image: confluentinc/cp-zookeeper:latest environment:
container_name: ${ZOOKEEPER_CONTAINER_NAME} ZOOKEEPER_CLIENT_PORT: 2181
hostname: zookeeper ZOOKEEPER_TICK_TIME: 2000
environment: ports:
ZOOKEEPER_CLIENT_PORT: 2181 - ${ZOOKEEPER_LOCAL_PORT}:2181
ZOOKEEPER_TICK_TIME: 2000 # volumes:
ports: # - ${DOCKER_DATA_PATH}/zookeeper:/var/lib/zookeeper
- ${ZOOKEEPER_LOCAL_PORT}:2181 networks:
# volumes: - gitlink_network
# - ${DOCKER_DATA_PATH}/zookeeper:/var/lib/zookeeper
networks: # See Also: https://hub.docker.com/r/confluentinc/cp-kafka
- gitlink_network kafka1:
restart: always image: confluentinc/cp-kafka:latest
container_name: ${KAFKA_CONTAINER_01_NAME}
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka hostname: kafka1
kafka1: depends_on:
image: confluentinc/cp-kafka:latest - zookeeper
container_name: ${KAFKA_CONTAINER_01_NAME} ports:
hostname: kafka1 - ${KAFKA_01_LOCAL_PORT}:29092
depends_on: # volumes:
- zookeeper # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/lib:/var/lib/kafka
ports: # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/logs:/var/logs/kafka
- ${KAFKA_01_LOCAL_PORT}:29092 # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/conf:/etc/kafka
# volumes: environment:
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/lib:/var/lib/kafka KAFKA_BROKER_ID: 1
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/logs:/var/logs/kafka KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/conf:/etc/kafka KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:${KAFKA_01_LOCAL_PORT}
environment: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_BROKER_ID: 1 KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:${KAFKA_01_LOCAL_PORT} networks:
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT - gitlink_network
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 # See Also: https://hub.docker.com/r/confluentinc/cp-kafka
networks: kafka2:
- gitlink_network image: confluentinc/cp-kafka:latest
restart: always container_name: ${KAFKA_CONTAINER_02_NAME}
hostname: kafka2
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka depends_on:
kafka2: - zookeeper
image: confluentinc/cp-kafka:latest ports:
container_name: ${KAFKA_CONTAINER_02_NAME} - ${KAFKA_02_LOCAL_PORT}:39092
hostname: kafka2 # volumes:
depends_on: # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/lib:/var/lib/kafka
- zookeeper # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/logs:/var/logs/kafka
ports: # - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/conf:/etc/kafka
- ${KAFKA_02_LOCAL_PORT}:39092 environment:
# volumes: KAFKA_BROKER_ID: 2
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/lib:/var/lib/kafka KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/logs:/var/logs/kafka KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,PLAINTEXT_HOST://localhost:${KAFKA_02_LOCAL_PORT}
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/conf:/etc/kafka KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
environment: KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_BROKER_ID: 2 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 networks:
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,PLAINTEXT_HOST://localhost:${KAFKA_02_LOCAL_PORT} - gitlink_network
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT gitlink-reader:
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 container_name: ${GNS_READER_CONTAINER_NAME}
networks: hostname: gitlink_reader
- gitlink_network image: gitlink/gns-reader:${GITLINK_NOTIFICATION_SYS_VERSION}
restart: always build:
context: ../
gitlink-reader: dockerfile: middleware/reader.Dockerfile
container_name: ${GNS_READER_CONTAINER_NAME} networks:
hostname: gitlink_reader - gitlink_network
image: gitlink/gns-reader:${GITLINK_NOTIFICATION_SYS_VERSION} environment:
build: - TZ=Asia/Shanghai
context: ../ volumes:
dockerfile: middleware/reader.Dockerfile - ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
networks: depends_on:
- gitlink_network - kafka1
environment: - kafka2
- TZ=Asia/Shanghai - redis
volumes: - mysql
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/ ports:
depends_on: - ${GNS_READER_LOCAL_PORT}:8081
- kafka1
- kafka2 gitlink-writer:
- redis container_name: ${GNS_WRITER_CONTAINER_NAME}
- mysql hostname: gitlink_writer
ports: image: gitlink/gns-writer:${GITLINK_NOTIFICATION_SYS_VERSION}
- ${GNS_READER_LOCAL_PORT}:8081 build:
restart: always context: ../
dockerfile: middleware/writer.Dockerfile
gitlink-writer: networks:
container_name: ${GNS_WRITER_CONTAINER_NAME} - gitlink_network
hostname: gitlink_writer environment:
image: gitlink/gns-writer:${GITLINK_NOTIFICATION_SYS_VERSION} - TZ=Asia/Shanghai
build: volumes:
context: ../ - ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
dockerfile: middleware/writer.Dockerfile depends_on:
networks: - kafka1
- gitlink_network - kafka2
environment: - redis
- TZ=Asia/Shanghai - mysql
volumes: ports:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/ - ${GNS_WRITER_LOCAL_PORT}:8082
depends_on:
- kafka1 gitlink-executor:
- kafka2 container_name: ${GNS_EXECUTOR_CONTAINER_NAME}
- redis hostname: gitlink_executor
- mysql image: gitlink/gns-executor:${GITLINK_NOTIFICATION_SYS_VERSION}
ports: build:
- ${GNS_WRITER_LOCAL_PORT}:8082 context: ../
restart: always dockerfile: middleware/executor.Dockerfile
networks:
gitlink-executor: - gitlink_network
container_name: ${GNS_EXECUTOR_CONTAINER_NAME} volumes:
hostname: gitlink_executor - ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
image: gitlink/gns-executor:${GITLINK_NOTIFICATION_SYS_VERSION} environment:
build: - TZ=Asia/Shanghai
context: ../ depends_on:
dockerfile: middleware/executor.Dockerfile - kafka1
networks: - kafka2
- gitlink_network - redis
volumes: - mysql
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/ ports:
environment: - ${GNS_EXECUTOR_LOCAL_PORT}:8083
- TZ=Asia/Shanghai
depends_on: networks:
- kafka1 gitlink_network:
- kafka2 driver: bridge
- redis name: gitlink_network
- mysql driver_opts:
ports: com.docker.network.enable_ipv6: "true"
- ${GNS_EXECUTOR_LOCAL_PORT}:8083
restart: always
networks:
gitlink_network:
driver: bridge
name: gitlink_network
driver_opts:
com.docker.network.enable_ipv6: "true"

0
middleware/start_docker_compose.sh Normal file → Executable file
View File

1
model/src/main/java/cn/org/gitlink/notification/model/dao/mapper/SysNotificationMapper.java
View File

@ -39,7 +39,6 @@ public interface SysNotificationMapper extends BaseMapper<SysNotification> {
List<SysNotification> getSysNotificationPageList(Page page, String orderBy, List<SysNotification> getSysNotificationPageList(Page page, String orderBy,
@Param("type") int type, @Param("type") int type,
@Param("sources") String sources,
@Param("platform") String platform, @Param("platform") String platform,
@Param("receiver") Integer receiver, @Param("receiver") Integer receiver,
@Param("status") Integer status); @Param("status") Integer status);

4
model/src/main/java/cn/org/gitlink/notification/model/service/notification/EmailJobsService.java
View File

@ -17,7 +17,7 @@ public interface EmailJobsService extends IService<EmailJob> {
* @Author: wanjia * @Author: wanjia
* @Date: 2021/9/13 * @Date: 2021/9/13
*/ */
boolean createEmailJob(NewEmailJobVo newEmailJobVo) throws Exception; boolean sendEmail(NewEmailJobVo newEmailJobVo) throws Exception;
/** /**
* 获取所有未处理邮件任务列表 * 获取所有未处理邮件任务列表
@ -42,6 +42,6 @@ public interface EmailJobsService extends IService<EmailJob> {
* @Author: wanjia * @Author: wanjia
* @Date: 2021/9/13 * @Date: 2021/9/13
*/ */
int markEmailJobAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) throws Exception; int markEmailJobsAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) throws Exception;
} }

3
model/src/main/java/cn/org/gitlink/notification/model/service/notification/EmailSendRecordsService.java
View File

@ -3,6 +3,7 @@ package cn.org.gitlink.notification.model.service.notification;
import cn.org.gitlink.notification.model.dao.entity.EmailSendRecord; import cn.org.gitlink.notification.model.dao.entity.EmailSendRecord;
import com.baomidou.mybatisplus.extension.service.IService; import com.baomidou.mybatisplus.extension.service.IService;
import java.util.Date;
import java.util.List; import java.util.List;
public interface EmailSendRecordsService extends IService<EmailSendRecord> { public interface EmailSendRecordsService extends IService<EmailSendRecord> {
@ -16,7 +17,7 @@ public interface EmailSendRecordsService extends IService<EmailSendRecord> {
* @Author: wanjia * @Author: wanjia
* @Date: 2021/9/13 * @Date: 2021/9/13
*/ */
boolean createEmailSendRecords(String platform, String emails, Integer jobId) throws Exception; boolean newEmailSendRecords(String platform, String emails, Integer jobId) throws Exception;
/** /**
* 获取发送记录列表 * 获取发送记录列表

9
model/src/main/java/cn/org/gitlink/notification/model/service/notification/SysNotificationService.java
View File

@ -36,17 +36,16 @@ public interface SysNotificationService extends IService<SysNotification> {
/** /**
* 获取消息列表 * 获取消息列表
* *
* @param type 类型 -1 全部 1 系统消息、2 @我
* @param page 页码
* @param size 页大小
* @param platform 平台编号 * @param platform 平台编号
* @param receiver 消息接收者 * @param receiver 消息接收者
* @param status 状态 -1 全部、1 未读 2 已读 * @param status 状态 -1 全部、1 未读 2 已读
* @param type 类型 -1 全部 1 系统消息、2 @我
* @param sources
* @param page 页码
* @param size 页大小
* @return * @return
*/ */
Page<SysNotification> getNotification(String platform, Integer receiver, Integer status, Integer type, String sources, Integer page, Integer size) throws Exception; Page<SysNotification> getNotification(String platform, Integer receiver, Integer status, Integer type, Integer page, Integer size) throws Exception;
/** /**
* @Description: 批量删除系统消息 * @Description: 批量删除系统消息

6
model/src/main/java/cn/org/gitlink/notification/model/service/notification/impl/EmailJobsServiceImpl.java
View File

@ -11,11 +11,11 @@ import org.springframework.stereotype.Service;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@Service(value = "EmailJobsServiceImpl") @Service
public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob> implements EmailJobsService { public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob> implements EmailJobsService {
@Override @Override
public boolean createEmailJob(NewEmailJobVo newEmailJobVo) { public boolean sendEmail(NewEmailJobVo newEmailJobVo) {
String platform = newEmailJobVo.getPlatform(); String platform = newEmailJobVo.getPlatform();
EmailJob emailJob = new EmailJob(); EmailJob emailJob = new EmailJob();
BeanUtils.copyProperties(newEmailJobVo, emailJob); BeanUtils.copyProperties(newEmailJobVo, emailJob);
@ -28,7 +28,7 @@ public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob>
} }
@Override @Override
public int markEmailJobAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) { public int markEmailJobsAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) throws Exception {
EmailJob emailJob = new EmailJob(); EmailJob emailJob = new EmailJob();
emailJob.setId(emailJobId); emailJob.setId(emailJobId);
emailJob.setDispatchedAt(dispatchedAt); emailJob.setDispatchedAt(dispatchedAt);

4
model/src/main/java/cn/org/gitlink/notification/model/service/notification/impl/EmailSendRecordsServiceImpl.java
View File

@ -11,12 +11,12 @@ import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.List; import java.util.List;
@Service(value = "EmailSendRecordsServiceImpl") @Service
public class EmailSendRecordsServiceImpl extends ServiceImpl<EmailSendRecordsMapper, EmailSendRecord> implements EmailSendRecordsService { public class EmailSendRecordsServiceImpl extends ServiceImpl<EmailSendRecordsMapper, EmailSendRecord> implements EmailSendRecordsService {
@Override @Override
@Transactional @Transactional
public boolean createEmailSendRecords(String platform, String emails, Integer jobId) { public boolean newEmailSendRecords(String platform, String emails, Integer jobId) {
List<String> list = Arrays.asList(emails.split(",")); List<String> list = Arrays.asList(emails.split(","));
List<EmailSendRecord> emailSendRecordList = new ArrayList<EmailSendRecord>(); List<EmailSendRecord> emailSendRecordList = new ArrayList<EmailSendRecord>();
for (String email : list){ for (String email : list){

18
model/src/main/java/cn/org/gitlink/notification/model/service/notification/impl/SysNotificationServiceImpl.java
View File

@ -28,7 +28,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
@Override @Override
@Transactional @Transactional
public boolean sendNotification(NewSysNotificationVo newSysNotificationVo) { public boolean sendNotification(NewSysNotificationVo newSysNotificationVo) throws Exception {
List<SysNotification> sysNotificationList = new ArrayList<>(); List<SysNotification> sysNotificationList = new ArrayList<>();
List<String> list = Arrays.asList(newSysNotificationVo.getReceivers().split(",")); List<String> list = Arrays.asList(newSysNotificationVo.getReceivers().split(","));
for (String receiver : list) { for (String receiver : list) {
@ -51,7 +51,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
} }
@Override @Override
public int markNotificationAs(String platform, Integer receiver, String notificationIds, Integer status, Integer type) { public int markNotificationAs(String platform, Integer receiver, String notificationIds, Integer status, Integer type) throws Exception {
int count = baseMapper.updateStatusByNotificationId(platform, receiver, notificationIds, status, type); int count = baseMapper.updateStatusByNotificationId(platform, receiver, notificationIds, status, type);
if (count > 0) { if (count > 0) {
this.delUserCache(platform, receiver); this.delUserCache(platform, receiver);
@ -60,7 +60,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
} }
@Override @Override
public int getNotificationCount(String platform, Integer receiver, Integer type, Integer status) { public int getNotificationCount(String platform, Integer receiver, Integer type, Integer status) throws Exception {
String cacheKey = cacheKeyForCount(platform, receiver, type, status); String cacheKey = cacheKeyForCount(platform, receiver, type, status);
Object foundResult = this.redisUtil.get(cacheKey); Object foundResult = this.redisUtil.get(cacheKey);
if (foundResult != null) { if (foundResult != null) {
@ -73,8 +73,8 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
@Override @Override
public Page<SysNotification> getNotification(String platform, Integer receiver, Integer status, Integer type, String sources, Integer page, Integer size) { public Page<SysNotification> getNotification(String platform, Integer receiver, Integer status, Integer type, Integer page, Integer size) throws Exception {
String cacheKey = cacheKeyForPage(platform, receiver, type, sources, status, page, size); String cacheKey = cacheKeyForPage(platform, receiver, type, status, page, size);
Object foundResult = this.redisUtil.get(cacheKey); Object foundResult = this.redisUtil.get(cacheKey);
if (foundResult != null) { if (foundResult != null) {
return (Page<SysNotification>) foundResult; return (Page<SysNotification>) foundResult;
@ -82,7 +82,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
Page<SysNotification> pageItem = new Page<SysNotification>(page, size); Page<SysNotification> pageItem = new Page<SysNotification>(page, size);
List<SysNotification> sysNotificationList = baseMapper.getSysNotificationPageList( List<SysNotification> sysNotificationList = baseMapper.getSysNotificationPageList(
pageItem, "", type, sources, platform, receiver, status pageItem, "", type, platform, receiver, status
); );
pageItem.setRecords(sysNotificationList); pageItem.setRecords(sysNotificationList);
this.redisUtil.set(cacheKey, pageItem); this.redisUtil.set(cacheKey, pageItem);
@ -90,7 +90,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
} }
@Override @Override
public int deleteNotifications(String platform, Integer receiver, String notificationIds, Integer type) { public int deleteNotifications(String platform, Integer receiver, String notificationIds, Integer type) throws Exception {
int count = baseMapper.deleteNotificationByIds(platform, receiver, notificationIds, type); int count = baseMapper.deleteNotificationByIds(platform, receiver, notificationIds, type);
if (count > 0) { if (count > 0) {
this.delUserCache(platform, receiver); this.delUserCache(platform, receiver);
@ -107,8 +107,8 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
return String.format("%s#T%s#S%s#Count", cachePrefixForPlatform(platform, receiver), type, status); return String.format("%s#T%s#S%s#Count", cachePrefixForPlatform(platform, receiver), type, status);
} }
private static String cacheKeyForPage(String platform, Integer receiver, Integer type, String sources, Integer status, Integer page, Integer size) { private static String cacheKeyForPage(String platform, Integer receiver, Integer type, Integer status, Integer page, Integer size) {
return String.format("%s#T%s#%s#S%s#P%s_S%s", cachePrefixForPlatform(platform, receiver), type, sources, status, page, size); return String.format("%s#T%s#S%s#P%s_S%s", cachePrefixForPlatform(platform, receiver), type, status, page, size);
} }
private static String cachePrefixForPlatform(String platform, Integer receiver) { private static String cachePrefixForPlatform(String platform, Integer receiver) {

6
model/src/main/resources/mapper/notification/SysNotificationMapper.xml
View File

@ -202,12 +202,6 @@
<if test="status != -1"> <if test="status != -1">
and status = #{status} and status = #{status}
</if> </if>
<if test="sources != null and sources != ''">
and source in
<foreach collection="sources.split(',')" item="sources" open="(" separator="," close=")">
#{sources}
</foreach>
</if>
ORDER BY id DESC ORDER BY id DESC
</select> </select>
<update id="deleteNotificationByIds"> <update id="deleteNotificationByIds">

6
model/src/test/java/cn/org/gitlink/notification/model/service/notification/EmailServiceTest.java
View File

@ -33,7 +33,7 @@ public class EmailServiceTest {
emailJobVo.setContent("contentTest"); emailJobVo.setContent("contentTest");
//添加发送邮件任务 //添加发送邮件任务
try { try {
Assert.isTrue(emailJobsService.createEmailJob(emailJobVo), "done"); Assert.isTrue(emailJobsService.sendEmail(emailJobVo), "done");
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
@ -49,7 +49,7 @@ public class EmailServiceTest {
try { try {
List<EmailJob> emailJobs = emailJobsService.getEmailJobsByDispatchedStatus("gitlink", -1, 10); List<EmailJob> emailJobs = emailJobsService.getEmailJobsByDispatchedStatus("gitlink", -1, 10);
for (EmailJob emailJob : emailJobs){ for (EmailJob emailJob : emailJobs){
emailSendRecordsService.createEmailSendRecords("gitlink", emailJob.getEmails(), emailJob.getId()); emailSendRecordsService.newEmailSendRecords("gitlink", emailJob.getEmails(), emailJob.getId());
} }
//发送邮件成功后，更新邮件发送记录 //发送邮件成功后，更新邮件发送记录
// int count = emailSendRecordsService.markEmailSendRecordsAs("gitlink", 1, new Date(), 1); // int count = emailSendRecordsService.markEmailSendRecordsAs("gitlink", 1, new Date(), 1);
@ -58,7 +58,7 @@ public class EmailServiceTest {
e.printStackTrace(); e.printStackTrace();
} }
try { try {
int count = emailJobsService.markEmailJobAs("gitlink", 1, new Date(), 1); int count = emailJobsService.markEmailJobsAs("gitlink", 1, new Date(), 1);
Assert.isTrue(count > 0, "update status success"); Assert.isTrue(count > 0, "update status success");
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();

2
model/src/test/java/cn/org/gitlink/notification/model/service/notification/ServiceTests.java
View File

@ -28,7 +28,7 @@ public class ServiceTests {
@Test @Test
public void testSysNotificationService() throws Exception { public void testSysNotificationService() throws Exception {
int i = sysNotificationService.getNotificationCount("gitlink", 234,1,1); int i = sysNotificationService.getNotificationCount("gitlink", 234,1,1);
Page<SysNotification> sysNotificationPage = sysNotificationService.getNotification("gitlink", 100,1,1, "IssueChanged", 1,20); Page<SysNotification> sysNotificationPage = sysNotificationService.getNotification("gitlink", 100,1,1,1,20);
NewSysNotificationVo newSysNotificationVo = new NewSysNotificationVo(); NewSysNotificationVo newSysNotificationVo = new NewSysNotificationVo();
newSysNotificationVo.setSender(1); newSysNotificationVo.setSender(1);
newSysNotificationVo.setReceivers("7,8"); newSysNotificationVo.setReceivers("7,8");

49
reader/src/main/java/cn/org/gitlink/notification/reader/controller/NotificationController.java
View File

@ -4,7 +4,6 @@ import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
import cn.org.gitlink.notification.common.response.DataPacketUtil; import cn.org.gitlink.notification.common.response.DataPacketUtil;
import cn.org.gitlink.notification.common.response.ResponseData; import cn.org.gitlink.notification.common.response.ResponseData;
import cn.org.gitlink.notification.common.utils.RedisUtil; import cn.org.gitlink.notification.common.utils.RedisUtil;
import cn.org.gitlink.notification.common.utils.ValidatorUtils;
import cn.org.gitlink.notification.model.dao.entity.vo.ReceiverNotificationCountVo; import cn.org.gitlink.notification.model.dao.entity.vo.ReceiverNotificationCountVo;
import cn.org.gitlink.notification.model.dao.entity.vo.ReceiverNotificationListVo; import cn.org.gitlink.notification.model.dao.entity.vo.ReceiverNotificationListVo;
import cn.org.gitlink.notification.model.service.notification.SysNotificationService; import cn.org.gitlink.notification.model.service.notification.SysNotificationService;
@ -20,6 +19,17 @@ import org.springframework.web.bind.annotation.*;
@RestController @RestController
@RequestMapping(value = "/gns/notification") @RequestMapping(value = "/gns/notification")
public class NotificationController { public class NotificationController {
//未读消息类型
private static final Integer NOTIFICATION_TYPE_ALL = -1; //全部未读消息
private static final Integer NOTIFICATION_TYPE_SYS = 1; //系统消息
private static final Integer NOTIFICATION_TYPE_ATME = 2; //@我的消息
//是否已读状态
private static final Integer STATUS_OPTION_ALL = -1; //所有信息类型，包括未读、已读
private static final Integer STATUS_OPTION_UNREAD = 1; //未读信息
private static final Integer STATUS_OPTION_READ = 2; //已读信息
@Autowired @Autowired
private RedisUtil redisUtil; private RedisUtil redisUtil;
@ -52,9 +62,6 @@ public class NotificationController {
@ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1") @ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1")
@RequestParam(name = "type", required = false, defaultValue = "-1") Integer type, @RequestParam(name = "type", required = false, defaultValue = "-1") Integer type,
@ApiParam(value = "消息来源")
@RequestParam(name = "sources", required = false) String sources,
@ApiParam(value = "页码：值为-1时（默认值），不开启分页；", required = false, defaultValue = "-1") @ApiParam(value = "页码：值为-1时（默认值），不开启分页；", required = false, defaultValue = "-1")
@RequestParam(name = "page", required = false, defaultValue = "-1") Integer page, @RequestParam(name = "page", required = false, defaultValue = "-1") Integer page,
@ -62,10 +69,7 @@ public class NotificationController {
@RequestParam(name = "size", required = false, defaultValue = "20") Integer size) { @RequestParam(name = "size", required = false, defaultValue = "20") Integer size) {
//参数合法性验证 //参数合法性验证
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform); ResponseData jsonFailResult = validatePlatformCodeAndReceiver(platform, receiver);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateReceiver(receiver);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateNotificationType(type); jsonFailResult = validateNotificationType(type);
@ -80,18 +84,18 @@ public class NotificationController {
try { try {
//获取全部未读消息总数 //获取全部未读消息总数
notificationListVo.setTotalUnreadNotifiationCount(notificationService.getNotificationCount(platform, receiver, NotificationSystemConstant.NOTIFICATION_TYPE_ALL, NotificationSystemConstant.STATUS_OPTION_UNREAD)); notificationListVo.setTotalUnreadNotifiationCount(notificationService.getNotificationCount(platform, receiver, NOTIFICATION_TYPE_ALL, STATUS_OPTION_UNREAD));
//获取未读系统消息总数 //获取未读系统消息总数
notificationListVo.setUnreadNotificationCount(notificationService.getNotificationCount(platform, receiver, NotificationSystemConstant.NOTIFICATION_TYPE_SYS, NotificationSystemConstant.STATUS_OPTION_UNREAD)); notificationListVo.setUnreadNotificationCount(notificationService.getNotificationCount(platform, receiver, NOTIFICATION_TYPE_SYS, STATUS_OPTION_UNREAD));
//获取未读@我消息总数 //获取未读@我消息总数
notificationListVo.setUnreadAtMeCount(notificationService.getNotificationCount(platform, receiver, NotificationSystemConstant.NOTIFICATION_TYPE_ATME, NotificationSystemConstant.STATUS_OPTION_UNREAD)); notificationListVo.setUnreadAtMeCount(notificationService.getNotificationCount(platform, receiver, NOTIFICATION_TYPE_ATME, STATUS_OPTION_UNREAD));
if (page == -1) { if (page == -1) {
page = 1; page = 1;
} }
//分页的数据 //分页的数据
Page foundPage = notificationService.getNotification(platform, receiver, status, type, sources, page, size); Page foundPage = notificationService.getNotification(platform, receiver, status, type, page, size);
if (foundPage != null) { if (foundPage != null) {
notificationListVo.setPageNum(foundPage.getCurrent()); notificationListVo.setPageNum(foundPage.getCurrent());
notificationListVo.setPageSize(foundPage.getSize()); notificationListVo.setPageSize(foundPage.getSize());
@ -126,10 +130,7 @@ public class NotificationController {
@ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1") @ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1")
@RequestParam(name = "type", required = false, defaultValue = "-1") Integer type) { @RequestParam(name = "type", required = false, defaultValue = "-1") Integer type) {
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform); ResponseData jsonFailResult = validatePlatformCodeAndReceiver(platform, receiver);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateReceiver(receiver);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateNotificationType(type); jsonFailResult = validateNotificationType(type);
@ -139,7 +140,7 @@ public class NotificationController {
ReceiverNotificationCountVo countVo = new ReceiverNotificationCountVo(); ReceiverNotificationCountVo countVo = new ReceiverNotificationCountVo();
countVo.setReceiver(receiver); countVo.setReceiver(receiver);
countVo.setType(type); countVo.setType(type);
countVo.setUnreadNotification(notificationService.getNotificationCount(platform, receiver, type, NotificationSystemConstant.STATUS_OPTION_UNREAD)); countVo.setUnreadNotification(notificationService.getNotificationCount(platform, receiver, type, STATUS_OPTION_UNREAD));
return DataPacketUtil.jsonSuccessResult(countVo); return DataPacketUtil.jsonSuccessResult(countVo);
} catch (Exception e) { } catch (Exception e) {
logger.error(e); logger.error(e);
@ -153,10 +154,18 @@ public class NotificationController {
/** /**
* 验证 platform 和 receiver 的合法性 * 验证 platform 和 receiver 的合法性
* *
* @param platform
* @param receiver * @param receiver
* @return * @return
*/ */
private ResponseData validateReceiver(Integer receiver) { private ResponseData validatePlatformCodeAndReceiver(String platform, Integer receiver) {
//验证 {platform} 参数合法性，以判断请求来源
if (!NotificationSystemConstant.PLATFORM_CODE_MAP.containsKey(platform)) {
logger.debug("\t 输入参数 {platform} 的值 {" + platform + "} 无效");
return DataPacketUtil.jsonFailResult("{platform} 参数非法");
}
//验证 {receiver} 参数合法性 //验证 {receiver} 参数合法性
if (receiver <= 0 || receiver >= Integer.MAX_VALUE) { if (receiver <= 0 || receiver >= Integer.MAX_VALUE) {
logger.debug("\t 输入参数 {receiver} 的值 {" + receiver + "} 超出约定范围"); logger.debug("\t 输入参数 {receiver} 的值 {" + receiver + "} 超出约定范围");
@ -173,7 +182,7 @@ public class NotificationController {
* @return * @return
*/ */
private ResponseData validateNotificationType(Integer type) { private ResponseData validateNotificationType(Integer type) {
if (type == NotificationSystemConstant.NOTIFICATION_TYPE_ALL || type == NotificationSystemConstant.NOTIFICATION_TYPE_SYS || type == NotificationSystemConstant.NOTIFICATION_TYPE_ATME) { if (type == NOTIFICATION_TYPE_ALL || type == NOTIFICATION_TYPE_SYS || type == NOTIFICATION_TYPE_ATME) {
return null; return null;
} else { } else {
logger.debug("\t 输入参数 {type} 的值 {" + type + "} 超出约定范围"); logger.debug("\t 输入参数 {type} 的值 {" + type + "} 超出约定范围");
@ -188,7 +197,7 @@ public class NotificationController {
* @return * @return
*/ */
private ResponseData validateStatusParams(Integer status) { private ResponseData validateStatusParams(Integer status) {
if (status == NotificationSystemConstant.STATUS_OPTION_ALL || status == NotificationSystemConstant.STATUS_OPTION_READ || status == NotificationSystemConstant.STATUS_OPTION_UNREAD) { if (status == STATUS_OPTION_ALL || status == STATUS_OPTION_READ || status == STATUS_OPTION_UNREAD) {
return null; return null;
} else { } else {
logger.debug("\t输入参数 {status} 的值 {" + status + "}超出约定范围"); logger.debug("\t输入参数 {status} 的值 {" + status + "}超出约定范围");

6
reader/src/main/resources/application.yml.example
View File

@ -18,6 +18,12 @@ spring:
min-idle: 0 min-idle: 0
timeout: 1000 timeout: 1000
kafka:
producer:
bootstrap_servers: kafka1:9092,kafka2:9092
retries: 5
batch_size: 16384
datasource: datasource:
driver-class-name: com.mysql.jdbc.Driver driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://mysql:3306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true&useSSL=false url: jdbc:mysql://mysql:3306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true&useSSL=false

43
writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java
View File

@ -1,5 +1,6 @@
package cn.org.gitlink.notification.writer.controller; package cn.org.gitlink.notification.writer.controller;
import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
import cn.org.gitlink.notification.common.response.DataPacketUtil; import cn.org.gitlink.notification.common.response.DataPacketUtil;
import cn.org.gitlink.notification.common.response.ResponseData; import cn.org.gitlink.notification.common.response.ResponseData;
import cn.org.gitlink.notification.common.utils.KafkaUtil; import cn.org.gitlink.notification.common.utils.KafkaUtil;
@ -9,21 +10,24 @@ import cn.org.gitlink.notification.model.dao.entity.vo.NewEmailJobVo;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam; import io.swagger.annotations.ApiParam;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException; import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.validation.BindingResult; import org.springframework.validation.BindingResult;
import org.springframework.validation.annotation.Validated; import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import java.util.Arrays;
import java.util.Map; import java.util.Map;
@RestController @RestController
@RequestMapping(value = "/gns/email") @RequestMapping(value = "/gns/email")
@Configuration
public class EmailJobsController { public class EmailJobsController {
private Logger logger = LogManager.getLogger(EmailJobsController.class); private Logger logger = LogManager.getLogger(EmailJobsController.class);
@ -31,19 +35,30 @@ public class EmailJobsController {
@Value("${spring.kafka.producer.topic_email}") @Value("${spring.kafka.producer.topic_email}")
private String gitlinkEmailTopic; private String gitlinkEmailTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired @Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil; private KafkaUtil kafkaUtil;
@ApiOperation("发送邮件任务") @ApiOperation("发送邮件任务")
@PostMapping(path = "/{platform}") @RequestMapping(path = "/{platform}", method = RequestMethod.POST)
@ResponseBody @ResponseBody
public ResponseData sendEmail(@ApiParam(value = "平台编码", required = true) public ResponseData sendEmail(@ApiParam(value = "平台编码", required = true)
@PathVariable(name = "platform") String platform, @PathVariable(name = "platform") String platform,
@Validated @RequestBody NewEmailJobParamsVo newEmailJobParamsVo){ @Validated @RequestBody NewEmailJobParamsVo newEmailJobParamsVo,
BindingResult bindingResult){
//参数合法性验证 //参数合法性验证
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform); Map<String, String> errors = ValidatorUtils.buildValidationErrorMessageMap(bindingResult);
if (!errors.isEmpty()) {
return DataPacketUtil.jsonFailResult(errors);
}
ResponseData jsonFailResult = validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
//platform和NewEmailJobParamsVo拼装 //platform和NewEmailJobParamsVo拼装
@ -57,6 +72,7 @@ public class EmailJobsController {
newEmailJobVo.setPlatform(platform); newEmailJobVo.setPlatform(platform);
try { try {
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkEmailTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkEmailTopic, JSONObject.toJSONString(newEmailJobVo)); kafkaUtil.sendMessage(gitlinkEmailTopic, JSONObject.toJSONString(newEmailJobVo));
return DataPacketUtil.jsonSuccessResult(); return DataPacketUtil.jsonSuccessResult();
} catch (Exception e) { } catch (Exception e) {
@ -64,4 +80,21 @@ public class EmailJobsController {
return DataPacketUtil.jsonFailResult(e.getMessage()); return DataPacketUtil.jsonFailResult(e.getMessage());
} }
} }
/**
* 验证 platform合法性
*
* @param platform
* @return
*/
private ResponseData validatePlatformCode(String platform) {
//验证 {platform} 参数合法性，以判断请求来源
if (!NotificationSystemConstant.PLATFORM_CODE_MAP.containsKey(platform)) {
logger.debug("\t 输入参数 {platform} 的值 {" + platform + "} 无效");
return DataPacketUtil.jsonFailResult("{platform} 参数非法");
}
return null;
}
} }

70
writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java
View File

@ -1,5 +1,6 @@
package cn.org.gitlink.notification.writer.controller; package cn.org.gitlink.notification.writer.controller;
import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
import cn.org.gitlink.notification.common.response.DataPacketUtil; import cn.org.gitlink.notification.common.response.DataPacketUtil;
import cn.org.gitlink.notification.common.response.ResponseData; import cn.org.gitlink.notification.common.response.ResponseData;
import cn.org.gitlink.notification.common.utils.KafkaUtil; import cn.org.gitlink.notification.common.utils.KafkaUtil;
@ -12,46 +13,59 @@ import cn.org.gitlink.notification.model.service.notification.SysNotificationSer
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam; import io.swagger.annotations.ApiParam;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException; import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.validation.BindingResult; import org.springframework.validation.BindingResult;
import org.springframework.validation.annotation.Validated; import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import java.util.Arrays;
import java.util.Map; import java.util.Map;
@RestController @RestController
@RequestMapping(value = "/gns/notification") @RequestMapping(value = "/gns/notification")
@Configuration
public class NotificationController { public class NotificationController {
@Value("${spring.kafka.producer.topic}") @Value("${spring.kafka.producer.topic}")
private String gitlinkNotificationTopic; private String gitlinkNotificationTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired @Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil; private KafkaUtil kafkaUtil;
@Autowired @Autowired
@Qualifier(value = "SysNotificationServiceImpl")
private SysNotificationService sysNotificationService; private SysNotificationService sysNotificationService;
private Logger logger = LogManager.getLogger(NotificationController.class); private Logger logger = LogManager.getLogger(NotificationController.class);
@ApiOperation("添加系统消息") @ApiOperation("添加系统消息")
@PostMapping(path = "/{platform}") @RequestMapping(path = "/{platform}", method = RequestMethod.POST)
@ResponseBody @ResponseBody
public ResponseData sendNotification(@ApiParam(value = "平台编码", required = true) public ResponseData sendNotification(@ApiParam(value = "平台编码", required = true)
@PathVariable(name = "platform") String platform, @PathVariable(name = "platform") String platform,
@Validated @RequestBody NewSysNotificationParamsVo newSysNotificationParamsVo) { @Validated @RequestBody NewSysNotificationParamsVo newSysNotificationParamsVo,
//platform校验 BindingResult bindingResult) {
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform);
//参数合法性验证
Map<String, String> errors = ValidatorUtils.buildValidationErrorMessageMap(bindingResult);
if (!errors.isEmpty()) {
return DataPacketUtil.jsonFailResult(errors);
}
ResponseData jsonFailResult = validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
//platform和NewSysNotificationParamsVo拼装 //platform和NewSysNotificationParamsVo拼装
@ -65,6 +79,7 @@ public class NotificationController {
newSysNotificationVo.setPlatform(platform); newSysNotificationVo.setPlatform(platform);
try { try {
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNotificationTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkNotificationTopic, JSONObject.toJSONString(newSysNotificationVo)); kafkaUtil.sendMessage(gitlinkNotificationTopic, JSONObject.toJSONString(newSysNotificationVo));
return DataPacketUtil.jsonSuccessResult(); return DataPacketUtil.jsonSuccessResult();
} catch (Exception e) { } catch (Exception e) {
@ -75,15 +90,21 @@ public class NotificationController {
} }
@ApiOperation("改变系统消息状态") @ApiOperation("改变系统消息状态")
@PutMapping(path = "/{platform}") @RequestMapping(path = "/{platform}", method = RequestMethod.PUT)
@ResponseBody @ResponseBody
public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true) public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true)
@PathVariable(name = "platform") String platform, @PathVariable(name = "platform") String platform,
@Validated @RequestBody UpdateNotificationStatusParamsVo updateNotificationStatusParamsVo) { @Validated @RequestBody UpdateNotificationStatusParamsVo updateNotificationStatusParamsVo,
BindingResult bindingResult) {
//参数合法性验证 //参数合法性验证
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform); Map<String, String> errors = ValidatorUtils.buildValidationErrorMessageMap(bindingResult);
if (!errors.isEmpty()) {
return DataPacketUtil.jsonFailResult(errors);
}
ResponseData jsonFailResult = validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
try { try {
@ -100,15 +121,21 @@ public class NotificationController {
} }
@ApiOperation("删除系统消息状态") @ApiOperation("删除系统消息状态")
@DeleteMapping(path = "/{platform}") @RequestMapping(path = "/{platform}", method = RequestMethod.DELETE)
@ResponseBody @ResponseBody
public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true) public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true)
@PathVariable(name = "platform") String platform, @PathVariable(name = "platform") String platform,
@Validated @RequestBody DeleteNotificationsVo deleteNotificationsVo) { @Validated @RequestBody DeleteNotificationsVo deleteNotificationsVo,
BindingResult bindingResult) {
//参数合法性验证 //参数合法性验证
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform); Map<String, String> errors = ValidatorUtils.buildValidationErrorMessageMap(bindingResult);
if (!errors.isEmpty()) {
return DataPacketUtil.jsonFailResult(errors);
}
ResponseData jsonFailResult = validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult; if (jsonFailResult != null) return jsonFailResult;
try { try {
@ -119,4 +146,21 @@ public class NotificationController {
return DataPacketUtil.jsonFailResult(e.getMessage()); return DataPacketUtil.jsonFailResult(e.getMessage());
} }
} }
/**
* 验证 platform合法性
*
* @param platform
* @return
*/
private ResponseData validatePlatformCode(String platform) {
//验证 {platform} 参数合法性，以判断请求来源
if (!NotificationSystemConstant.PLATFORM_CODE_MAP.containsKey(platform)) {
logger.debug("\t 输入参数 {platform} 的值 {" + platform + "} 无效");
return DataPacketUtil.jsonFailResult("{platform} 参数非法");
}
return null;
}
} }

2
writer/src/main/resources/application.yml.example
View File

@ -14,8 +14,6 @@ spring:
partitions: 3 partitions: 3
topic: topic-gitlink-notification topic: topic-gitlink-notification
topic_email: topic-gitlink-email topic_email: topic-gitlink-email
# 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
topics: ${spring.kafka.producer.topic}, ${spring.kafka.producer.topic_email}
redis: redis:
database: 0 database: 0