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

Merge pull request 'EmailJobsListener注释KafkaListener注解' (#80) from wanjia9506/gitlink-notification-system:dev_gitlink_model into master

This commit is contained in:
baladiwei 2021-09-15 16:28:15 +08:00
parent 4a59fb481e 2903ac29c7
commit 3b5c11bb10
9 changed files with 67 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,14 +25,13 @@ public class EmailService {
private EmailSendRecordsService emailSendRecordsService;
/**
* @Description: 处理邮件发送任务，根据emails添加到邮件发送记录表中。
*
* @Param platform 平台编码
* @Param dispatchNumber 待处理发送任务列表数量
* @return: void
* @Author: wanjia
* @Date: 2021/9/15
*/
* @Description: 处理邮件发送任务，根据emails添加到邮件发送记录表中。
* @Param platform 平台编码
* @Param dispatchNumber 待处理发送任务列表数量
* @return: void
* @Author: wanjia
* @Date: 2021/9/15
*/
public void DispatchEmailJobs(String platform, Integer dispatchNumber) {
//获取指定数量待处理列表
List<EmailJob> emailJobList = new ArrayList<>();
@ -52,7 +51,7 @@ public class EmailService {
}
//EmailJob分配成功，更新状态
try {
emailJobsService.markEmailJobAs(platform, emailJob.getId(), new Date(), flag ? 1 : 2);
emailJobsService.markEmailJobsAs(platform, emailJob.getId(), new Date(), flag ? 1 : 2);
} catch (Exception e) {
logger.error("更新EmailJob状态失败，email_job_id: " + emailJob.getId() + "\n" + e);
}
@ -60,15 +59,14 @@ public class EmailService {
}
/**
* @Description: 发送邮件
*
* @Param platform 平台编码
* @Param sentNumber 一次发送数量
* @return: void
* @Author: wanjia
* @Date: 2021/9/15
*/
public void sendEmail(String platform, Integer sentNumber){
* @Description: 发送邮件
* @Param platform 平台编码
* @Param sentNumber 一次发送数量
* @return: void
* @Author: wanjia
* @Date: 2021/9/15
*/
public void sendEmail(String platform, Integer sentNumber) {
List<EmailSendRecord> emailSendRecordList = new ArrayList<>();
try {
emailSendRecordList = emailSendRecordsService.getRecordsByStatus(platform, -1, sentNumber);
@ -77,17 +75,23 @@ public class EmailService {
}
Boolean flag = null;
for (EmailSendRecord emailSendRecord : emailSendRecordList){
List<EmailSendRecord> updateEmailSendRecordList = new ArrayList<>();
EmailSendRecord sentEmailSendRecord = new EmailSendRecord();
for (EmailSendRecord unSentEmailSendRecord : emailSendRecordList) {
//todo 发邮件
// 取到一条 emailSendRecord 调用发邮件的Util，返回发送结果赋值给flag
//更新emailSendRecord状态
try {
emailSendRecordsService.markEmailSendRecordAs(platform, emailSendRecord.getId(), new Date(), flag ? 1 : 2);
} catch (Exception e){
logger.error("更新EmailSendRecord状态失败，email_send_record_id: " + emailSendRecord.getId() + "\n" + e);
}
sentEmailSendRecord.setId(unSentEmailSendRecord.getId());
sentEmailSendRecord.setSentAt(new Date());
sentEmailSendRecord.setStatus(flag ? 1 : 2);
updateEmailSendRecordList.add(sentEmailSendRecord);
}
//更新emailSendRecord状态
try {
emailSendRecordsService.markEmailSendRecordsAs(platform, updateEmailSendRecordList);
} catch (Exception e) {
logger.error(e);
}
}
}

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

@ -14,7 +14,7 @@ import org.springframework.stereotype.Component;
@Component
@Configuration
//todo 邮件的topics和groupId待处理
@KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group_id}")
//@KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group_id}")
public class EmailJobsListener {
private Logger logger = LogManager.getLogger(EmailJobsListener.class);

2
model/src/main/java/cn/org/gitlink/notification/model/dao/mapper/EmailSendRecordsMapper.java
View File

@ -23,4 +23,6 @@ public interface EmailSendRecordsMapper extends BaseMapper<EmailSendRecord> {
int insertEmailSendRecordBatch(@Param("platform") String platform,@Param("list") List<EmailSendRecord> emailSendRecordList);
List<EmailSendRecord> getRecordsByStatus(@Param("platform") String platform, @Param("status") Integer status, @Param("size") Integer sentNumber);
int updateEmailSendRecordsBatch(@Param("platform") String platform, @Param("list") List<EmailSendRecord> emailSendRecordList);
}

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

@ -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;
}

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

@ -42,6 +42,6 @@ public interface EmailSendRecordsService extends IService<EmailSendRecord> {
* @Author: wanjia
* @Date: 2021/9/13
*/
int markEmailSendRecordAs(String platform, Integer emailSendRecordId, Date sentAt, Integer status) throws Exception;
int markEmailSendRecordsAs(String platform, List<EmailSendRecord> emailSendRecordList) throws Exception;
}

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

@ -31,7 +31,7 @@ public class EmailJobsServiceImpl extends ServiceImpl<EmailJobsMapper, EmailJob>
}
@Override
public int markEmailJobAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) throws Exception {
public int markEmailJobsAs(String platform, Integer emailJobId, Date dispatchedAt, Integer dispatchedStatus) throws Exception {
EmailJob emailJob = new EmailJob();
emailJob.setId(emailJobId);
emailJob.setDispatchedAt(dispatchedAt);

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

@ -31,16 +31,12 @@ public class EmailSendRecordsServiceImpl extends ServiceImpl<EmailSendRecordsMap
}
@Override
public int markEmailSendRecordAs(String platform, Integer emailSendRecordId, Date sentAt, Integer status) throws Exception {
EmailSendRecord emailSendRecord = new EmailSendRecord();
emailSendRecord.setId(emailSendRecordId);
emailSendRecord.setSentAt(sentAt);
emailSendRecord.setStatus(status);
return baseMapper.updateByPrimaryKeySelective(platform, emailSendRecord);
public int markEmailSendRecordsAs(String platform, List<EmailSendRecord> emailSendRecordList){
return baseMapper.updateEmailSendRecordsBatch(platform, emailSendRecordList);
}
@Override
public List<EmailSendRecord> getRecordsByStatus(String platform, Integer status, Integer sentNumber) throws Exception {
public List<EmailSendRecord> getRecordsByStatus(String platform, Integer status, Integer sentNumber){
return baseMapper.getRecordsByStatus(platform, status, sentNumber);
}
}

25
model/src/main/resources/mapper/notification/EmailSendRecordsMapper.xml
View File

@ -121,4 +121,29 @@
select * from ${platform}_email_send_records
where status = #{status} limit #{size}
</select>
<update id="updateEmailSendRecordsBatch" parameterType="list">
update ${platform}_email_send_records
<trim prefix="set" suffixOverrides=",">
<trim prefix="sentAt =case" suffix="end,">
<foreach collection="list" item="i" index="index">
<if test="i.sentAt!=null">
when id=#{i.id} then #{i.sentAt}
</if>
</foreach>
</trim>
<trim prefix=" status =case" suffix="end,">
<foreach collection="list" item="i" index="index">
<if test="i.status!=null">
when id=#{i.id} then #{i.status}
</if>
</foreach>
</trim>
</trim>
where
<foreach collection="list" separator="or" item="i" index="index" >
id=#{i.id}
</foreach>
</update>
</mapper>

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

@ -52,13 +52,13 @@ public class EmailServiceTest {
emailSendRecordsService.newEmailSendRecords("gitlink", emailJob.getEmails(), emailJob.getId());
}
//发送邮件成功后，更新邮件发送记录
int count = emailSendRecordsService.markEmailSendRecordAs("gitlink", 1, new Date(), 1);
Assert.isTrue(count > 0, "update email_send_record status success");
// int count = emailSendRecordsService.markEmailSendRecordsAs("gitlink", 1, new Date(), 1);
// Assert.isTrue(count > 0, "update email_send_record status success");
} catch (Exception e){
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();