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 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 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` 绝对路径到本地磁盘

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

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

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;
/**
* 消息系统常量
*
* @description:
* @author: wanjia
* @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_HEHUI = "hehui"; //hehui平台
public static final String PLATFORM_CODE_OSREDM = "osredm"; //红山开源平台
public static final Map<String, String> PLATFORM_CODE_MAP = new HashMap<String, String>() {
{
put("gitlink", PLATFORM_CODE_GITLINK);
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.LoggerFactory;
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.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* web请求异常全局处理类，在使用的模块内继承此类，然后加上ControllerAdvice注解，example:
@ -44,24 +39,4 @@ public class ExceptionInterceptor {
logger.error(logTemplate, request.getRequestURI(), e);
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.List;
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 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.atomic.AtomicReference;
import java.util.stream.Collectors;
@ -26,7 +29,7 @@ import java.util.stream.Collectors;
* @date 2021-09-07
*/
@Component(value = "GitlinkKafkaUtil")
@Component
public class KafkaUtil {
private Logger logger = LogManager.getLogger(KafkaUtil.class);
@ -34,15 +37,6 @@ public class KafkaUtil {
@Value("${spring.kafka.producer.bootstrap_servers:#{null}}")
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;
@Autowired
@ -52,32 +46,16 @@ public class KafkaUtil {
* 初始化AdminClient
* '@PostConstruct该注解被用来修饰一个非静态的void（）方法。
* 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。
* PostConstruct在构造函数之后执行，init（）方法之前执行。ls
* PostConstruct在构造函数之后执行，init（）方法之前执行。
*/
@PostConstruct
private void initAdminClient() {
Map<String, Object> props = new HashMap<>(1);
// 如果当前配置信息内没有kafka producer相关配置则不对adminClient做初始化
if (kafkaServer == null) {
return;
}
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
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) {

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;
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.FieldError;
@ -25,19 +21,4 @@ public class ValidatorUtils {
}
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`),
KEY `index_on_email_and_status` (`email`,`status`),
KEY `index_on_status` (`status`)
) 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;
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;

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 字段保存额外信息
ALTER TABLE gitlink_sys_notification ADD source varchar(250) 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;
ALTER TABLE gitlink_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）';

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 字段保存额外信息
ALTER TABLE hehui_sys_notification ADD source varchar(250) 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;
ALTER TABLE hehui_sys_notification ADD extra TEXT NULL COMMENT '额外信息（备用字段）';

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

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;
import cn.org.gitlink.notification.common.constant.NotificationSystemConstant;
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.EmailSendRecord;
@ -22,6 +21,16 @@ public class EmailService {
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
private EmailJobsService emailJobsService;
@ -45,7 +54,7 @@ public class EmailService {
//获取指定数量待处理列表
List<EmailJob> emailJobList = new ArrayList<>();
try {
emailJobList = emailJobsService.getEmailJobsByDispatchedStatus(platform, NotificationSystemConstant.EMAIL_JOB_NOT_DISPATCHED, dispatchNumber);
emailJobList = emailJobsService.getEmailJobsByDispatchedStatus(platform, EMAIL_JOB_NOT_DISPATCHED, dispatchNumber);
} catch (Exception e) {
logger.error("获取未处理邮件任务列表失败：\n" + e);
}
@ -54,13 +63,13 @@ public class EmailService {
Boolean flag = null;
for (EmailJob emailJob : emailJobList) {
try {
flag = emailSendRecordsService.createEmailSendRecords(platform, emailJob.getEmails(), emailJob.getId());
flag = emailSendRecordsService.newEmailSendRecords(platform, emailJob.getEmails(), emailJob.getId());
} catch (Exception e) {
logger.error("处理EmailJob失败，email_job_id: " + emailJob.getId() + "\n" + e);
}
//EmailJob分配成功，更新状态
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) {
logger.error("更新EmailJob状态失败，email_job_id: " + emailJob.getId() + "\n" + e);
}
@ -81,7 +90,7 @@ public class EmailService {
//获取待发送列表
List<EmailSendRecord> emailSendRecordList = new ArrayList<>();
try {
emailSendRecordList = emailSendRecordsService.getRecordsByStatus(platform, NotificationSystemConstant.EMAIL_UNSENT_RECORD, sentNumber);
emailSendRecordList = emailSendRecordsService.getRecordsByStatus(platform, EMAIL_UNSENT_RECORD, sentNumber);
} catch (Exception e) {
logger.error("获取未发送邮件列表失败：\n" + e);
}
@ -94,9 +103,8 @@ public class EmailService {
emailUtils.sendMail(unSentEmailSendRecord.getSubject(), unSentEmailSendRecord.getEmail(), unSentEmailSendRecord.getContent());
flag = true;
unSentEmailSendRecord.setSentAt(new Date());
unSentEmailSendRecord.setStatus(flag ? NotificationSystemConstant.EMAIL_SENT_SUCCESS : NotificationSystemConstant.EMAIL_SENT_FAIL);
} catch (Exception e) {
unSentEmailSendRecord.setStatus(NotificationSystemConstant.EMAIL_SENT_FAIL);
unSentEmailSendRecord.setStatus(flag ? EMAIL_SENT_SUCCESS : EMAIL_SENT_FAIL);
} catch (MessagingException 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.service.notification.EmailJobsService;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Arrays;
@Component
@Configuration
@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}")
private String gitlinkNewEmailRemindTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil;
@KafkaHandler
public void messageHandler(String message) {
try {
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 (flag){
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNewEmailRemindTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkNewEmailRemindTopic, JSONObject.toJSONString(newEmailJobVo));
}
} catch (Exception e) {

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

@ -13,8 +13,6 @@ spring:
replication_factor: 1
partitions: 3
topic_new_email_remind: topic-gitlink-new-email-remind
# 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
topics: ${spring.kafka.producer.topic_new_email_remind}
consumer:
bootstrap_servers: kafka1:9092,kafka2:9092
group_id: group-gitlink-notification
@ -28,7 +26,7 @@ spring:
topic_email: topic-gitlink-email
topic_new_email_remind: topic-gitlink-new-email-remind
listener:
concurrency: 1
concurrency: 3
ack-mode: record
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'
services:
mysql:
image: mysql:${MYSQL_VERSION}
container_name: ${MYSQL_CONTAINER_NAME}
hostname: mysql
environment:
- MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
- MYSQL_USER=${MYSQL_USER}
- MYSQL_PASSWORD=${MYSQL_PASSWORD}
- MYSQL_DATABASE=${MYSQL_DATABASE}
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/mysql:/var/lib/mysql
- ${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}/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
ports:
- ${MYSQL_LOCAL_PORT}:3306
networks:
- gitlink_network
restart: always
redis:
image: redis:${REDIS_VERSION}
container_name: ${REDIS_CONTAINER_NAME}
hostname: redis
volumes:
- ${DOCKER_DATA_PATH}/redis/data:/data
- ${DOCKER_DATA_PATH}/redis/logs:/logs
environment:
- TZ=Asia/Shanghai
ports:
- ${REDIS_LOCAL_PORT}:6379
networks:
- gitlink_network
restart: always
# See Also: https://hub.docker.com/r/confluentinc/cp-zookeeper
zookeeper:
image: confluentinc/cp-zookeeper:latest
container_name: ${ZOOKEEPER_CONTAINER_NAME}
hostname: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
ports:
- ${ZOOKEEPER_LOCAL_PORT}:2181
# volumes:
# - ${DOCKER_DATA_PATH}/zookeeper:/var/lib/zookeeper
networks:
- gitlink_network
restart: always
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka
kafka1:
image: confluentinc/cp-kafka:latest
container_name: ${KAFKA_CONTAINER_01_NAME}
hostname: kafka1
depends_on:
- zookeeper
ports:
- ${KAFKA_01_LOCAL_PORT}:29092
# volumes:
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/lib:/var/lib/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/logs:/var/logs/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/conf:/etc/kafka
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:${KAFKA_01_LOCAL_PORT}
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
networks:
- gitlink_network
restart: always
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka
kafka2:
image: confluentinc/cp-kafka:latest
container_name: ${KAFKA_CONTAINER_02_NAME}
hostname: kafka2
depends_on:
- zookeeper
ports:
- ${KAFKA_02_LOCAL_PORT}:39092
# volumes:
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/lib:/var/lib/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/logs:/var/logs/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/conf:/etc/kafka
environment:
KAFKA_BROKER_ID: 2
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,PLAINTEXT_HOST://localhost:${KAFKA_02_LOCAL_PORT}
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
networks:
- gitlink_network
restart: always
gitlink-reader:
container_name: ${GNS_READER_CONTAINER_NAME}
hostname: gitlink_reader
image: gitlink/gns-reader:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/reader.Dockerfile
networks:
- gitlink_network
environment:
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_READER_LOCAL_PORT}:8081
restart: always
gitlink-writer:
container_name: ${GNS_WRITER_CONTAINER_NAME}
hostname: gitlink_writer
image: gitlink/gns-writer:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/writer.Dockerfile
networks:
- gitlink_network
environment:
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_WRITER_LOCAL_PORT}:8082
restart: always
gitlink-executor:
container_name: ${GNS_EXECUTOR_CONTAINER_NAME}
hostname: gitlink_executor
image: gitlink/gns-executor:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/executor.Dockerfile
networks:
- gitlink_network
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
environment:
- TZ=Asia/Shanghai
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_EXECUTOR_LOCAL_PORT}:8083
restart: always
networks:
gitlink_network:
driver: bridge
name: gitlink_network
driver_opts:
com.docker.network.enable_ipv6: "true"
version: '3'
services:
mysql:
image: mysql:${MYSQL_VERSION}
container_name: ${MYSQL_CONTAINER_NAME}
hostname: mysql
environment:
- MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
- MYSQL_USER=${MYSQL_USER}
- MYSQL_PASSWORD=${MYSQL_PASSWORD}
- MYSQL_DATABASE=${MYSQL_DATABASE}
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/mysql:/var/lib/mysql
- ${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}/gns-email.sql:/docker-entrypoint-initdb.d/0003.sql
command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
ports:
- ${MYSQL_LOCAL_PORT}:3306
networks:
- gitlink_network
redis:
image: redis:${REDIS_VERSION}
container_name: ${REDIS_CONTAINER_NAME}
hostname: redis
volumes:
- ${DOCKER_DATA_PATH}/redis/data:/data
- ${DOCKER_DATA_PATH}/redis/logs:/logs
environment:
- TZ=Asia/Shanghai
ports:
- ${REDIS_LOCAL_PORT}:6379
networks:
- gitlink_network
# See Also: https://hub.docker.com/r/confluentinc/cp-zookeeper
zookeeper:
image: confluentinc/cp-zookeeper:latest
container_name: ${ZOOKEEPER_CONTAINER_NAME}
hostname: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
ports:
- ${ZOOKEEPER_LOCAL_PORT}:2181
# volumes:
# - ${DOCKER_DATA_PATH}/zookeeper:/var/lib/zookeeper
networks:
- gitlink_network
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka
kafka1:
image: confluentinc/cp-kafka:latest
container_name: ${KAFKA_CONTAINER_01_NAME}
hostname: kafka1
depends_on:
- zookeeper
ports:
- ${KAFKA_01_LOCAL_PORT}:29092
# volumes:
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/lib:/var/lib/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/logs:/var/logs/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_01_NAME}/conf:/etc/kafka
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:${KAFKA_01_LOCAL_PORT}
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
networks:
- gitlink_network
# See Also: https://hub.docker.com/r/confluentinc/cp-kafka
kafka2:
image: confluentinc/cp-kafka:latest
container_name: ${KAFKA_CONTAINER_02_NAME}
hostname: kafka2
depends_on:
- zookeeper
ports:
- ${KAFKA_02_LOCAL_PORT}:39092
# volumes:
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/lib:/var/lib/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/logs:/var/logs/kafka
# - ${DOCKER_DATA_PATH}/kafka/${KAFKA_CONTAINER_02_NAME}/conf:/etc/kafka
environment:
KAFKA_BROKER_ID: 2
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,PLAINTEXT_HOST://localhost:${KAFKA_02_LOCAL_PORT}
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
networks:
- gitlink_network
gitlink-reader:
container_name: ${GNS_READER_CONTAINER_NAME}
hostname: gitlink_reader
image: gitlink/gns-reader:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/reader.Dockerfile
networks:
- gitlink_network
environment:
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_READER_LOCAL_PORT}:8081
gitlink-writer:
container_name: ${GNS_WRITER_CONTAINER_NAME}
hostname: gitlink_writer
image: gitlink/gns-writer:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/writer.Dockerfile
networks:
- gitlink_network
environment:
- TZ=Asia/Shanghai
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_WRITER_LOCAL_PORT}:8082
gitlink-executor:
container_name: ${GNS_EXECUTOR_CONTAINER_NAME}
hostname: gitlink_executor
image: gitlink/gns-executor:${GITLINK_NOTIFICATION_SYS_VERSION}
build:
context: ../
dockerfile: middleware/executor.Dockerfile
networks:
- gitlink_network
volumes:
- ${DOCKER_DATA_PATH}/gitlink/:/data/logs/
environment:
- TZ=Asia/Shanghai
depends_on:
- kafka1
- kafka2
- redis
- mysql
ports:
- ${GNS_EXECUTOR_LOCAL_PORT}:8083
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,
@Param("type") int type,
@Param("sources") String sources,
@Param("platform") String platform,
@Param("receiver") Integer receiver,
@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
* @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
* @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 com.baomidou.mybatisplus.extension.service.IService;
import java.util.Date;
import java.util.List;
public interface EmailSendRecordsService extends IService<EmailSendRecord> {
@ -16,7 +17,7 @@ public interface EmailSendRecordsService extends IService<EmailSendRecord> {
* @Author: wanjia
* @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 receiver 消息接收者
* @param status 状态 -1 全部、1 未读 2 已读
* @param type 类型 -1 全部 1 系统消息、2 @我
* @param sources
* @param page 页码
* @param size 页大小
* @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: 批量删除系统消息

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.List;
@Service(value = "EmailJobsServiceImpl")
@Service
public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob> implements EmailJobsService {
@Override
public boolean createEmailJob(NewEmailJobVo newEmailJobVo) {
public boolean sendEmail(NewEmailJobVo newEmailJobVo) {
String platform = newEmailJobVo.getPlatform();
EmailJob emailJob = new EmailJob();
BeanUtils.copyProperties(newEmailJobVo, emailJob);
@ -28,7 +28,7 @@ public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob>
}
@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.setId(emailJobId);
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.List;
@Service(value = "EmailSendRecordsServiceImpl")
@Service
public class EmailSendRecordsServiceImpl extends ServiceImpl<EmailSendRecordsMapper, EmailSendRecord> implements EmailSendRecordsService {
@Override
@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<EmailSendRecord> emailSendRecordList = new ArrayList<EmailSendRecord>();
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
@Transactional
public boolean sendNotification(NewSysNotificationVo newSysNotificationVo) {
public boolean sendNotification(NewSysNotificationVo newSysNotificationVo) throws Exception {
List<SysNotification> sysNotificationList = new ArrayList<>();
List<String> list = Arrays.asList(newSysNotificationVo.getReceivers().split(","));
for (String receiver : list) {
@ -51,7 +51,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
}
@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);
if (count > 0) {
this.delUserCache(platform, receiver);
@ -60,7 +60,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
}
@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);
Object foundResult = this.redisUtil.get(cacheKey);
if (foundResult != null) {
@ -73,8 +73,8 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
@Override
public Page<SysNotification> getNotification(String platform, Integer receiver, Integer status, Integer type, String sources, Integer page, Integer size) {
String cacheKey = cacheKeyForPage(platform, receiver, type, sources, status, page, 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, status, page, size);
Object foundResult = this.redisUtil.get(cacheKey);
if (foundResult != null) {
return (Page<SysNotification>) foundResult;
@ -82,7 +82,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
Page<SysNotification> pageItem = new Page<SysNotification>(page, size);
List<SysNotification> sysNotificationList = baseMapper.getSysNotificationPageList(
pageItem, "", type, sources, platform, receiver, status
pageItem, "", type, platform, receiver, status
);
pageItem.setRecords(sysNotificationList);
this.redisUtil.set(cacheKey, pageItem);
@ -90,7 +90,7 @@ public class SysNotificationServiceImpl extends ServiceImpl<SysNotificationMappe
}
@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);
if (count > 0) {
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);
}
private static String cacheKeyForPage(String platform, Integer receiver, Integer type, String sources, 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);
private static String cacheKeyForPage(String platform, Integer receiver, Integer type, Integer status, Integer page, Integer 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) {

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

@ -202,12 +202,6 @@
<if test="status != -1">
and status = #{status}
</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
</select>
<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");
//添加发送邮件任务
try {
Assert.isTrue(emailJobsService.createEmailJob(emailJobVo), "done");
Assert.isTrue(emailJobsService.sendEmail(emailJobVo), "done");
} catch (Exception e) {
e.printStackTrace();
}
@ -49,7 +49,7 @@ public class EmailServiceTest {
try {
List<EmailJob> emailJobs = emailJobsService.getEmailJobsByDispatchedStatus("gitlink", -1, 10);
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);
@ -58,7 +58,7 @@ public class EmailServiceTest {
e.printStackTrace();
}
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");
} catch (Exception e) {
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
public void testSysNotificationService() throws Exception {
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.setSender(1);
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.ResponseData;
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.ReceiverNotificationListVo;
import cn.org.gitlink.notification.model.service.notification.SysNotificationService;
@ -20,6 +19,17 @@ import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping(value = "/gns/notification")
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
private RedisUtil redisUtil;
@ -52,9 +62,6 @@ public class NotificationController {
@ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1")
@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")
@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) {
//参数合法性验证
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateReceiver(receiver);
ResponseData jsonFailResult = validatePlatformCodeAndReceiver(platform, receiver);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateNotificationType(type);
@ -80,18 +84,18 @@ public class NotificationController {
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) {
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) {
notificationListVo.setPageNum(foundPage.getCurrent());
notificationListVo.setPageSize(foundPage.getSize());
@ -126,10 +130,7 @@ public class NotificationController {
@ApiParam(value = "消息类型：值为-1时，获取全部信息；值为1时，获取系统消息；值为2时，获取@我消息", defaultValue = "-1")
@RequestParam(name = "type", required = false, defaultValue = "-1") Integer type) {
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateReceiver(receiver);
ResponseData jsonFailResult = validatePlatformCodeAndReceiver(platform, receiver);
if (jsonFailResult != null) return jsonFailResult;
jsonFailResult = validateNotificationType(type);
@ -139,7 +140,7 @@ public class NotificationController {
ReceiverNotificationCountVo countVo = new ReceiverNotificationCountVo();
countVo.setReceiver(receiver);
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);
} catch (Exception e) {
logger.error(e);
@ -153,10 +154,18 @@ public class NotificationController {
/**
* 验证 platform 和 receiver 的合法性
*
* @param platform
* @param receiver
* @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} 参数合法性
if (receiver <= 0 || receiver >= Integer.MAX_VALUE) {
logger.debug("\t 输入参数 {receiver} 的值 {" + receiver + "} 超出约定范围");
@ -173,7 +182,7 @@ public class NotificationController {
* @return
*/
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;
} else {
logger.debug("\t 输入参数 {type} 的值 {" + type + "} 超出约定范围");
@ -188,7 +197,7 @@ public class NotificationController {
* @return
*/
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;
} else {
logger.debug("\t输入参数 {status} 的值 {" + status + "}超出约定范围");

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

@ -18,6 +18,12 @@ spring:
min-idle: 0
timeout: 1000
kafka:
producer:
bootstrap_servers: kafka1:9092,kafka2:9092
retries: 5
batch_size: 16384
datasource:
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

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;
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 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 io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.validation.BindingResult;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import java.util.Arrays;
import java.util.Map;
@RestController
@RequestMapping(value = "/gns/email")
@Configuration
public class EmailJobsController {
private Logger logger = LogManager.getLogger(EmailJobsController.class);
@ -31,19 +35,30 @@ public class EmailJobsController {
@Value("${spring.kafka.producer.topic_email}")
private String gitlinkEmailTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil;
@ApiOperation("发送邮件任务")
@PostMapping(path = "/{platform}")
@RequestMapping(path = "/{platform}", method = RequestMethod.POST)
@ResponseBody
public ResponseData sendEmail(@ApiParam(value = "平台编码", required = true)
@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;
//platform和NewEmailJobParamsVo拼装
@ -57,6 +72,7 @@ public class EmailJobsController {
newEmailJobVo.setPlatform(platform);
try {
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkEmailTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkEmailTopic, JSONObject.toJSONString(newEmailJobVo));
return DataPacketUtil.jsonSuccessResult();
} catch (Exception e) {
@ -64,4 +80,21 @@ public class EmailJobsController {
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;
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 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 io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.validation.BindingResult;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import java.util.Arrays;
import java.util.Map;
@RestController
@RequestMapping(value = "/gns/notification")
@Configuration
public class NotificationController {
@Value("${spring.kafka.producer.topic}")
private String gitlinkNotificationTopic;
@Value("${spring.kafka.producer.partitions}")
private Integer partitions;
@Value("${spring.kafka.producer.replication_factor}")
private Short replicationFactor;
@Autowired
@Qualifier(value = "GitlinkKafkaUtil")
private KafkaUtil kafkaUtil;
@Autowired
@Qualifier(value = "SysNotificationServiceImpl")
private SysNotificationService sysNotificationService;
private Logger logger = LogManager.getLogger(NotificationController.class);
@ApiOperation("添加系统消息")
@PostMapping(path = "/{platform}")
@RequestMapping(path = "/{platform}", method = RequestMethod.POST)
@ResponseBody
public ResponseData sendNotification(@ApiParam(value = "平台编码", required = true)
@PathVariable(name = "platform") String platform,
@Validated @RequestBody NewSysNotificationParamsVo newSysNotificationParamsVo) {
@Validated @RequestBody NewSysNotificationParamsVo newSysNotificationParamsVo,
//platform校验
ResponseData jsonFailResult = ValidatorUtils.validatePlatformCode(platform);
BindingResult bindingResult) {
//参数合法性验证
Map<String, String> errors = ValidatorUtils.buildValidationErrorMessageMap(bindingResult);
if (!errors.isEmpty()) {
return DataPacketUtil.jsonFailResult(errors);
}
ResponseData jsonFailResult = validatePlatformCode(platform);
if (jsonFailResult != null) return jsonFailResult;
//platform和NewSysNotificationParamsVo拼装
@ -65,6 +79,7 @@ public class NotificationController {
newSysNotificationVo.setPlatform(platform);
try {
kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNotificationTopic, partitions, replicationFactor)));
kafkaUtil.sendMessage(gitlinkNotificationTopic, JSONObject.toJSONString(newSysNotificationVo));
return DataPacketUtil.jsonSuccessResult();
} catch (Exception e) {
@ -75,15 +90,21 @@ public class NotificationController {
}
@ApiOperation("改变系统消息状态")
@PutMapping(path = "/{platform}")
@RequestMapping(path = "/{platform}", method = RequestMethod.PUT)
@ResponseBody
public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true)
@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;
try {
@ -100,15 +121,21 @@ public class NotificationController {
}
@ApiOperation("删除系统消息状态")
@DeleteMapping(path = "/{platform}")
@RequestMapping(path = "/{platform}", method = RequestMethod.DELETE)
@ResponseBody
public ResponseData changeNotificationStatus(@ApiParam(value = "平台编码", required = true)
@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;
try {
@ -119,4 +146,21 @@ public class NotificationController {
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
topic: topic-gitlink-notification
topic_email: topic-gitlink-email
# 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
topics: ${spring.kafka.producer.topic}, ${spring.kafka.producer.topic_email}
redis:
database: 0