kafka|反向匹配通知邮件

@@ -2,6 +2,7 @@ package com.fedex.connect.kafka.repository.base;
 
 import com.fedex.connect.common.dao.biz.CeInfoMapper;
 import com.fedex.connect.common.dao.biz.ConsignmentMapper;
+import com.fedex.connect.common.dao.biz.EmailMapper;
 import com.fedex.connect.common.dao.sys.KafkaStorageHistoryMapper;
 import com.fedex.connect.common.dao.sys.KafkaTemporaryStorageMapper;
 import com.fedex.connect.kafka.repository.dao.CeInfoMapperExt;
@@ -24,4 +25,6 @@ public class BaseRepository {
     protected KafkaTemporaryStorageMapperExt kafkaTemporaryStorageMapperExt;
     @Autowired
     protected UserMapperExt userMapperExt;
+    @Autowired
+    protected EmailMapper emailMapper;
 }
\ No newline at end of file

@@ -20,4 +20,9 @@ public interface UserMapperExt {
             "SELECT * FROM T_SYS_USER WHERE USER_UUID = #{uuid} AND ROWNUM <= 1" +
             "</script>")
     User findUserByUUid(@Param("uuid") String uuid);
+
+    @Select( "<script>" +
+            "SELECT * FROM T_SYS_USER WHERE ID = #{id}" +
+            "</script>")
+    User selectUserById(@Param("id") Long id);
 }
\ No newline at end of file

+package com.fedex.connect.kafka.repository.repo;
+
+import com.fedex.connect.common.model.biz.Email;
+
+public interface IEmailRepository {
+    void save(Email entity);
+}

@@ -4,4 +4,5 @@ import com.fedex.connect.common.model.sys.User;
 
 public interface IUserRepository {
     User findUserByUUid(String uuid);
+    User selectUserById(Long id);
 }

+package com.fedex.connect.kafka.repository.repo.impl;
+
+import com.fedex.connect.common.model.biz.Email;
+import com.fedex.connect.kafka.repository.base.BaseRepository;
+import com.fedex.connect.kafka.repository.repo.IEmailRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public class EmailRepositoryImpl extends BaseRepository implements IEmailRepository {
+    public void save(Email entity){
+        if(entity != null) {
+            if (entity.getId() == null || entity.getId() <= 0) {
+                emailMapper.insertSelective(entity);
+            } else if (entity.getId() != null && entity.getId() > 0){
+                emailMapper.updateByPrimaryKeySelective(entity);
+            }
+        }
+    }
+}

@@ -23,4 +23,9 @@ public class UserRepositoryImpl extends BaseRepository implements IUserRepositor
     public User findUserByUUid(String uuid) {
         return userMapperExt.findUserByUUid(uuid);
     }
+
+    @Override
+    public User selectUserById(Long id){
+        return userMapperExt.selectUserById(id);
+    }
 }

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.fedex.connect.common.dependencies.util.Utils;
 import com.fedex.connect.common.model.biz.CeInfo;
 import com.fedex.connect.common.model.biz.Consignment;
+import com.fedex.connect.common.model.biz.Email;
 import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
 import com.fedex.connect.kafka.data.dto.DmProcessResults;
 import com.fedex.connect.kafka.data.dto.dm.dm501.Dm501Consignments;
@@ -14,7 +15,6 @@ import com.fedex.connect.kafka.util.Dm501Util;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 import java.util.Objects;
@@ -55,11 +55,12 @@ public class Dm501ServiceImpl extends BaseService implements IDm501Service {
                 if (Objects.isNull(bizCeInfo)) {
                     CeInfo ceInfo = dm501Util.generateCeInfo(consignment501,kafKaTemporaryStorage.getSendTime());
                     //ce数据解析正常，进行后续操作
-                    Consignment consignment = dm501Util.generateConsignment(ceInfo);
+                    Email email = new Email();
+                    Consignment consignment = dm501Util.generateConsignment(ceInfo,email);
                     /**
                      * 保存ce信息、以及运单表信息
                      */
-                    dm501Util.saveCeInfoAndConsignment(ceInfo,consignment);
+                    dm501Util.saveCeInfoAndConsignment(ceInfo,consignment,email);
                 }
             }catch(Exception ex){
                 log.error("DM501消息处理异常：{}", ex.getMessage(),ex);

 package com.fedex.connect.kafka.util;
 
 import com.fedex.connect.common.dependencies.cache.CacheSystem;
+import com.fedex.connect.common.dependencies.enums.biz.EmailStatusEnum;
+import com.fedex.connect.common.dependencies.enums.biz.EmailTypeEnum;
 import com.fedex.connect.common.dependencies.util.AssignmentFieldUtils;
 import com.fedex.connect.common.dependencies.util.DateUtil;
 import com.fedex.connect.common.model.bi.DictionaryEntries;
 import com.fedex.connect.common.model.biz.CeInfo;
 import com.fedex.connect.common.model.biz.Consignment;
+import com.fedex.connect.common.model.biz.Email;
 import com.fedex.connect.common.model.sys.User;
 import com.fedex.connect.kafka.constants.Constant;
 import com.fedex.connect.kafka.data.dto.dm.dm501.Dm501ConAddresses;
 import com.fedex.connect.kafka.data.dto.dm.dm501.Dm501Consignments;
 import com.fedex.connect.kafka.repository.repo.ICeInfoRepository;
 import com.fedex.connect.kafka.repository.repo.IConsignmentRepository;
+import com.fedex.connect.kafka.repository.repo.IEmailRepository;
 import com.fedex.connect.kafka.repository.repo.IUserRepository;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
@@ -23,6 +27,7 @@ import java.math.BigDecimal;
 import java.util.Date;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 
 @Component
 public class Dm501Util {
@@ -35,6 +40,8 @@ public class Dm501Util {
     ICeInfoRepository ceInfoRepository;
     @Autowired
     IUserRepository userRepository;
+    @Autowired
+    IEmailRepository emailRepository;
 
     /**
      * @Author mt
@@ -197,7 +204,7 @@ public class Dm501Util {
       * @param ceInfo
       * @return com.fedex.connect.common.model.biz.Consignment
       */
-    public Consignment generateConsignment(CeInfo ceInfo) throws Exception{
+    public Consignment generateConsignment(CeInfo ceInfo,Email email) throws Exception{
         Consignment rsConsignment;
         /**
          * 根据运单号查找30天之内运单
@@ -206,10 +213,6 @@ public class Dm501Util {
         if(Objects.isNull(consignment)){
             Consignment resultConsignment = new Consignment();
             BeanUtils.copyProperties(ceInfo,resultConsignment);
-            //运单状态赋值为"待上传"
-//            DictionaryEntries consignmentDicEntries = cacheSystem.getDicConsignmentStatus(ConsignmentStatusEnum.CONSIGNMENT_STATUS_01.getCode());
-//            resultConsignment.setStatusCode(consignmentDicEntries.getCode());
-//            resultConsignment.setStatusName(consignmentDicEntries.getEnglishName());
             /**
              * 初始化运单表原产国、目的国
              */
@@ -228,6 +231,10 @@ public class Dm501Util {
             AssignmentFieldUtils.assignmentTableBaseField(resultConsignment);
             rsConsignment = resultConsignment;
         }else{
+            /**
+             * 初始化提醒邮件
+             */
+            this.initNotificationEmail(email,consignment,ceInfo);
             //ceInfo转换为consignment忽略字段
             String[] ignoreProperties = Constant.CE_CONSIGNMENT_COPY_IGNORE_PROP_KEYS.IGNORE_PROPERTIES;
             //字段拷贝
@@ -241,6 +248,31 @@ public class Dm501Util {
         return rsConsignment;
     }
 
+    private void initNotificationEmail(Email email,Consignment consignment,CeInfo ceInfo) throws Exception{
+        /**
+         * 初始化提醒发件人邮件
+         */
+        User userOld = userRepository.selectUserById(consignment.getSubmitterId());
+        if (StringUtils.isEmpty(consignment.getUserUuid()) && !userOld.getUserUuid().equals(ceInfo.getUserUuid())){
+            /**
+             * 初始化Email对象
+             */
+            email.setBizId(consignment.getId());
+            email.setBizCode(consignment.getConsignmentCode());
+            String emailAddress = Optional.ofNullable(ceInfo.getUserUuid())
+                    .map(userRepository::findUserByUUid)
+                    .map(user -> StringUtils.defaultIfEmpty(user.getEmail(), ceInfo.getShipperEmail()))
+                    .orElse(consignment.getShipperEmail());
+            email.setToAddress(emailAddress);
+            email.setSubject("");
+            email.setTypeName(EmailTypeEnum.NOTIFICATION_SENDER.getMsg());
+            email.setTypeCode(EmailTypeEnum.NOTIFICATION_SENDER.getCode());
+            email.setStatusName(EmailStatusEnum.PENDING.getMsg());
+            email.setStatusCode(EmailStatusEnum.PENDING.getCode());
+            AssignmentFieldUtils.assignmentTableBaseField(email);
+        }
+    }
+
     /**
       * @Author mt
       * @Description 初始化运单表原产国、目的国
@@ -297,10 +329,16 @@ public class Dm501Util {
      * @return void
      */
     @Transactional(rollbackFor = Exception.class)
-    public void saveCeInfoAndConsignment(CeInfo ceInfo, Consignment consignment){
+    public void saveCeInfoAndConsignment(CeInfo ceInfo, Consignment consignment,Email email){
         ceInfoRepository.save(ceInfo);
         //绑定ceInfo表Id
         consignment.setCeInfoId(ceInfo.getId());
         consignmentRepository.save(consignment);
+        /**
+         * 保存入库
+         */
+        if (email != null){
+            emailRepository.save(email);
+        }
     }
 }