feat(*): 重构发送消息

- 重构发送消息

feat(*): 重构发送消息
- 重构发送消息
张志伟
1 parent 385010d7
Showing 11 changed files with 185 additions and 167 deletions
fw-hermes-common/src/main/java/cn/fw/hermes/common/constant/Constant.java
fw-hermes-common/src/main/java/cn/fw/hermes/common/utils/StringUtils.java
fw-hermes-domain/src/main/java/cn/fw/hermes/domain/db/MessageHistory.java
fw-hermes-domain/src/main/java/cn/fw/hermes/domain/mqtt/Message.java
fw-hermes-server/src/main/kotlin/cn/fw/hermes/controller/common/XinGeDebugController.kt
fw-hermes-service/src/main/java/cn/fw/hermes/service/biz/BeanTransfer.java
fw-hermes-service/src/main/java/cn/fw/hermes/service/biz/MessageSender.java
fw-hermes-service/src/main/kotlin/cn/fw/hermes/service/biz/MessageSender.kt
fw-hermes-service/src/main/kotlin/cn/fw/hermes/service/biz/MsgPrepareBizService.kt
fw-hermes-service/src/main/kotlin/cn/fw/hermes/service/biz/XgPushService.kt
fw-hermes-service/src/main/java/cn/fw/hermes/service/emqx/MqttService.kt → fw-hermes-service/src/main/kotlin/cn/fw/hermes/service/emqx/MqttService.kt
@@ -2,24 +2,24 @@ package cn.fw.hermes.common.constant;
  
 import cn.hutool.core.util.StrUtil;
  
-import java.util.function.Function;
+import java.util.function.UnaryOperator;
  
 /**
  * @author 张志伟
  * @date 2018-03-29 11:40
  */
 public interface Constant {
-     String SUCCESS_MSG = "OK";
+    String SUCCESS_MSG = "OK";
     String PLATFORM_KEY = "x-appid";
     String TOPIC_CACHE_KEY = "sys:topic";
     String PLATFORM_CACHE_KEY = "sys:platform";
     String ACCOUNT_CACHE_KEY = "account:info";
     String DISTURB_SETTING_CACHE_KEY = "setting:disturb";
-    Function<String, String> RPC_EHR_CACHE_KEY = r -> StrUtil.format("rpc:ehr:{}", r);
+    UnaryOperator<String> RPC_EHR_CACHE_KEY = r -> StrUtil.format("rpc:ehr:{}", r);
     /**
      * 唯一id缓存key
      */
-    Function<String, String> UNIQUE_ID_CACHE_KEY = r -> StrUtil.format("hermes:unique:id:{}", r);
+    UnaryOperator<String> UNIQUE_ID_CACHE_KEY = r -> StrUtil.format("hermes:unique:id:{}", r);
     /**
      * 系统topic
      */
@@ -66,5 +66,5 @@ public interface Constant {
     /**
      * 无效信鸽token
      */
-    String inValidToken = "#";
+    String INVALID_TOKEN = "#";
 }
@@ -305,8 +305,7 @@ public final class StringUtils {
                 return input;
             }
             buffer.append(tail);
-        } catch (UnsupportedEncodingException e) {
-            e.printStackTrace();
+        } catch (UnsupportedEncodingException ignored) {
         }
         return buffer.toString();
     }
@@ -465,8 +464,7 @@ public final class StringUtils {
         int len = text.length();
         try {
             len = text.getBytes("GBK").length;//SQLServer数据库用的GBK编码
-        } catch (UnsupportedEncodingException e) {
-            e.printStackTrace();
+        } catch (UnsupportedEncodingException ignored) {
         }
         return len;
     }
@@ -483,7 +481,7 @@ public final class StringUtils {
         int pos = -1;
         int i = 0;
         while ((pos = data.indexOf(from, i)) != -1) {
-            buf.append(data.substring(i, pos)).append(to);
+            buf.append(data, i, pos).append(to);
             i = pos + from.length();
         }
         buf.append(data.substring(i));
@@ -543,4 +541,4 @@ public final class StringUtils {
         System.out.println(repeat("?", 10, ","));
         length("AAA中国（）111222bb");
     }
-}
 \ No newline at end of file
+}
@@ -50,13 +50,13 @@ public class MessageHistory {
     @TableField(typeHandler = StringListTypeHandler.class)
     private List<String> msgTopic;
     /**
-     * 目标账号
+     * 目标账号 [对应userId]
      */
     private String toAccount;
     /**
-     * 是否是待办消息
+     * 是否是需要信鸽推送的待办消息
      */
-    private Boolean td;
+    private Boolean xgTodo;
     /**
      * 发送方账号
      */
 package cn.fw.hermes.domain.mqtt;
  
-import cn.fw.hermes.domain.db.MessageHistory;
 import cn.fw.hermes.domain.mqtt.msgbody.MsgElement;
 import lombok.Getter;
  
-import java.time.ZoneOffset;
 import java.util.List;
  
 /**
@@ -58,17 +56,6 @@ public class Message {
      */
     private String remark;
  
-    public static Message with(MessageHistory history) {
-        Message msg = new Message();
-        msg.setMsgId(history.getId());
-        msg.setMsgSummary(history.getMsgSummary());
-        msg.setSender(history.getFromAccount());
-        msg.setReceiver(history.getToAccount());
-        msg.setSendTimestamp(history.getSendTime().toInstant(ZoneOffset.of("+8")).toEpochMilli());
-        msg.setMsgTimestamp(history.getCreateTime().getTime());
-        return msg;
-    }
-
  
     public void setMsgId(Long msgId) {
         this.msgId = msgId;
@@ -32,7 +32,7 @@ class XinGeDebugController(private val xinGeProperty: XinGeProperty, private val
     @ControllerMethod("信鸽推送测试消息")
     @GetMapping("/send/msg")
     fun sendMsg(@NotNull(message = "用户id不能为空") userId: Long?): Message<Void> {
-        userId?.let { xgPushService.sendData(it, AccountTypeEnum.B_USER) }
+        userId?.let { xgPushService.sendDebugData(it, AccountTypeEnum.B_USER) }
         return success()
     }
  
@@ -111,7 +111,7 @@ public final class BeanTransfer {
     public static MessageHistory transform2MsgHistory(MessageConditionDTO dto) {
         MessageHistory history = new MessageHistory();
         history.setId(dto.getId());
-        history.setTd(false);
+        history.setXgTodo(false);
         MessageBodyDTO firstMsgBody = dto.getMsgBodyList().get(0);
         MsgTypeEnum msgTypeEnum = MsgTypeEnum.ofValue(firstMsgBody.getType());
         String format = "[%s]";
@@ -138,8 +138,9 @@ public final class BeanTransfer {
                 history.setMsgSummary(String.format(format, "其他"));
                 String data = firstMsgBody.getData();
                 boolean isBackLog = data.indexOf(Constant.BACKLOG_MSG) > 0;
-                if (isBackLog) {
-                    history.setTd(true);
+                boolean hasApp = dto.getPlatforms().stream().anyMatch(t -> t.equals(PlatformTypeEnum.APP.getValue()));
+                if (isBackLog && hasApp) {
+                    history.setXgTodo(true);
                     history.setMsgSummary(String.format(format, "待办"));
                 }
                 break;
-package cn.fw.hermes.service.biz;
-
-import cn.fw.hermes.common.constant.Constant;
-import cn.fw.hermes.common.utils.DateUtil;
-import cn.fw.hermes.domain.db.MessageData;
-import cn.fw.hermes.domain.db.MessageHistory;
-import cn.fw.hermes.domain.enums.MsgStatusEnum;
-import cn.fw.hermes.domain.enums.MsgTypeEnum;
-import cn.fw.hermes.domain.mqtt.Message;
-import cn.fw.hermes.domain.mqtt.msgbody.*;
-import cn.fw.hermes.sdk.v2.enums.MsgState;
-import cn.fw.hermes.sdk.v2.result.MessageV2Result;
-import cn.fw.hermes.service.data.MessageDataService;
-import cn.fw.hermes.service.data.MessageHistoryService;
-import cn.fw.hermes.service.emqx.MqttService;
-import cn.fw.hermes.service.mq.MessageProducer;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-import org.springframework.util.CollectionUtils;
-
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.util.Date;
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-/**
- * 消息发送服务
- *
- * @author : kurisu
- * @version : 2.0
- * @desc : 消息发送服务
- * @date : 2023-07-21 15:11
- */
-@RequiredArgsConstructor
-@Service
-@Slf4j
-public class MessageSender {
-    private final MessageHistoryService messageHistoryService;
-    private final MessageDataService messageDataService;
-    private final MqttService mqttService;
-    private final MessageProducer messageProducer;
-    private final XgPushService xgPushService;
-
-    @Transactional(rollbackFor = Exception.class)
-    public void sendMsg(Long messageId) {
-        MessageHistory history = messageHistoryService.getById(messageId);
-        if (Objects.isNull(history)) {
-            return;
-        }
-        if (!MsgStatusEnum.WAITING.equals(history.getMsgStatus())) {
-            return;
-        }
-        List<MessageData> dataList = messageDataService.queryDataByMsgId(history.getId());
-        if (CollectionUtils.isEmpty(dataList)) {
-            return;
-        }
-        Pair<MsgStatusEnum, String> pair = sendMsg(history, dataList);
-        if (Objects.isNull(pair) || MsgStatusEnum.UPLOAD.equals(pair.getLeft())) {
-            return;
-        }
-        if (Boolean.TRUE.equals(history.getNeedResult())) {
-            MessageV2Result result = new MessageV2Result();
-            result.setMessageId(history.getId());
-            result.setMsgState(MsgState.FAIL.getValue());
-            result.setMsgStateDesc(history.getMsgStatusDesc());
-            result.setResultTime(DateUtil.toDate(LocalDateTime.now()));
-            messageProducer.send(result);
-        }
-    }
-
-    @Transactional(rollbackFor = Exception.class)
-    public Pair<MsgStatusEnum, String> sendMsg(MessageHistory history, List<MessageData> dataList) {
-        history.setSendTime(LocalDateTime.now());
-        history.setSendDay(LocalDate.now());
-        history.setCreateTime(new Date());
-        final Message msg = Message.with(history);
-        List<MsgElement> list = dataList.stream().map(r -> this.genElement(r, msg)).collect(Collectors.toList());
-        msg.setMsgBody(list);
-        List<String> msgTopics = history.getMsgTopic();
-        Pair<Boolean, String> pair = mqttService.senMsg(msgTopics, msg);
-        if (Objects.isNull(pair)) {
-            return null;
-        }
-        boolean isSuccess = Boolean.TRUE.equals(pair.getKey());
-        String desc = pair.getValue();
-        history.setMsgStatus(MsgStatusEnum.UPLOAD);
-        history.setMsgStatusDesc(desc);
-        if (!isSuccess) {
-            history.setMsgStatus(MsgStatusEnum.FAIL);
-            history.setSendSuccessTime(LocalDateTime.now());
-        }
-        messageHistoryService.saveOrUpdate(history);
-        return Pair.of(history.getMsgStatus(), desc);
-    }
-
-    private MsgElement genElement(MessageData msgData, Message msg) {
-        MsgTypeEnum msgTypeEnum = msgData.getType();
-        switch (msgTypeEnum) {
-            case TEXT:
-                return TextMsg.generate(msgData.getMsgId(), msgData.getText());
-            case IMAGE:
-                return ImageMsg.generate(msgData.getMsgId(), msgData.getImageUrl());
-            case VOICE:
-                return VoiceMsg.generate(msgData.getMsgId(), msgData.getVoiceUrl(), msgData.getDuration());
-            case VIDEO:
-                return VideoMsg.generate(msgData.getMsgId(), msgData.getDuration(), msgData.getVideoUrl());
-            case EMOJI:
-                return EmojiMsg.generate(msgData.getMsgId(), msgData.getEmojiIndex());
-            case LOCATION:
-                return LocationMsg.generate(msgData.getMsgId(), msgData.getText(), msgData.getLat(), msgData.getLng());
-            case CUSTOM:
-                String data = msgData.getData();
-                boolean isBackLog = data.indexOf(Constant.BACKLOG_MSG) > 0;
-                if (isBackLog) {
-                    msg.setRemark(Constant.BACKLOG_MSG);
-                }
-                return CustomMsg.generate(msgData.getMsgId(), data);
-            default:
-                break;
-        }
-        return null;
-    }
-}
+package cn.fw.hermes.service.biz
+
+import cn.fw.hermes.common.constant.Constant
+import cn.fw.hermes.common.utils.DateUtil
+import cn.fw.hermes.domain.db.MessageData
+import cn.fw.hermes.domain.db.MessageHistory
+import cn.fw.hermes.domain.enums.AccountTypeEnum
+import cn.fw.hermes.domain.enums.MsgStatusEnum
+import cn.fw.hermes.domain.enums.MsgTypeEnum
+import cn.fw.hermes.domain.enums.PlatformTypeEnum
+import cn.fw.hermes.domain.mqtt.Message
+import cn.fw.hermes.domain.mqtt.msgbody.*
+import cn.fw.hermes.sdk.v2.enums.MsgState
+import cn.fw.hermes.sdk.v2.result.MessageV2Result
+import cn.fw.hermes.service.data.DisturbSettingService
+import cn.fw.hermes.service.data.MessageDataService
+import cn.fw.hermes.service.data.MessageHistoryService
+import cn.fw.hermes.service.data.SysPlatformService
+import cn.fw.hermes.service.emqx.MqttService
+import cn.fw.hermes.service.mq.MessageProducer
+import org.apache.commons.lang3.tuple.Pair
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+import org.springframework.util.CollectionUtils
+import java.time.LocalDate
+import java.time.LocalDateTime
+import java.time.LocalTime
+import java.time.ZoneOffset
+import java.util.*
+import java.util.stream.Collectors
+
+/**
+ * 消息发送服务
+ *
+ * @author : kurisu
+ * @version : 2.0
+ * @desc : 消息发送服务
+ * @date : 2023-07-21 15:11
+ */
+@Service
+class MessageSender(
+    private val messageHistoryService: MessageHistoryService,
+    private val messageDataService: MessageDataService,
+    private val mqttService: MqttService,
+    private val messageProducer: MessageProducer,
+    private val xgPushService: XgPushService,
+    private val sysPlatformService: SysPlatformService,
+    private val disturbSettingService: DisturbSettingService
+) {
+
+    @Transactional(rollbackFor = [Exception::class])
+    fun sendMsg(messageId: Long?) {
+        val history = messageHistoryService.getById(messageId)
+        history?.takeIf { MsgStatusEnum.WAITING != it.msgStatus }?.run {
+            val dataList = messageDataService.queryDataByMsgId(history.id)
+            if (CollectionUtils.isEmpty(dataList)) {
+                return
+            }
+            val pair = sendMsg(history, dataList)
+            if (Objects.isNull(pair) || MsgStatusEnum.UPLOAD == pair?.left) {
+                return
+            }
+            if (history.needResult == true) {
+                val result = MessageV2Result()
+                result.setMessageId(history.id)
+                result.setMsgState(MsgState.FAIL.value)
+                result.setMsgStateDesc(history.msgStatusDesc)
+                result.setResultTime(DateUtil.toDate(LocalDateTime.now()))
+                messageProducer.send(result)
+            }
+        }
+    }
+
+    @Transactional(rollbackFor = [Exception::class])
+    fun sendMsg(history: MessageHistory, dataList: List<MessageData>): Pair<MsgStatusEnum, String>? {
+        val xgTodo = history.xgTodo
+        val userId = history.toAccount.toLong()
+        val platformCode = sysPlatformService.queryPlatforms().find { it.platformType == PlatformTypeEnum.APP }?.run { platformCode } ?: Constant.INVALID_TOKEN
+        val userDisturb = disturbSettingService.queryByUserPlatform(userId, platformCode)
+        val useMqtt = userDisturb?.takeIf { it.enabled == true }?.run { this@MessageSender.isTimeInRange(startTime, endTime) } ?: false
+        history.setSendTime(LocalDateTime.now())
+        history.setSendDay(LocalDate.now())
+        history.setCreateTime(Date())
+        if (useMqtt || !xgTodo) {
+            val msg = history.toMsg()
+            val list = dataList.stream().map { r: MessageData -> this.genElement(r, msg) }.collect(Collectors.toList())
+            msg.msgBody = list
+            val msgTopics = history.msgTopic
+            val pair = mqttService.senMsg(msgTopics, msg)
+            return pair?.let {
+                val isSuccess = true == it.key
+                val desc = it.value
+                history.setMsgStatus(MsgStatusEnum.UPLOAD)
+                history.setMsgStatusDesc(desc)
+                if (!isSuccess) {
+                    history.setMsgStatus(MsgStatusEnum.FAIL)
+                    history.setSendSuccessTime(LocalDateTime.now())
+                }
+                messageHistoryService.saveOrUpdate(history)
+                Pair.of(history.msgStatus, desc)
+            }
+        }
+
+        return xgPushService.sendMsg(userId, AccountTypeEnum.B_USER).run {
+            val sendSuccess = first == 0
+            history.setSendSuccessTime(LocalDateTime.now())
+            if (sendSuccess) {
+                history.setMsgStatus(MsgStatusEnum.SUCCESS)
+                history.setMsgStatusDesc(Constant.SUCCESS_MSG)
+                history.setXgId(second)
+            } else {
+                history.setMsgStatus(MsgStatusEnum.FAIL)
+                history.setMsgStatusDesc("$first:$second")
+            }
+            messageHistoryService.saveOrUpdate(history)
+
+            Pair.of(history.msgStatus, history.msgStatusDesc)
+        }
+    }
+
+    private fun genElement(msgData: MessageData, msg: Message): MsgElement? {
+        val msgTypeEnum = msgData.type
+        when (msgTypeEnum) {
+            MsgTypeEnum.TEXT -> return TextMsg.generate(msgData.msgId, msgData.text)
+            MsgTypeEnum.IMAGE -> return ImageMsg.generate(msgData.msgId, msgData.imageUrl)
+            MsgTypeEnum.VOICE -> return VoiceMsg.generate(msgData.msgId, msgData.voiceUrl, msgData.duration)
+            MsgTypeEnum.VIDEO -> return VideoMsg.generate(msgData.msgId, msgData.duration, msgData.videoUrl)
+            MsgTypeEnum.EMOJI -> return EmojiMsg.generate(msgData.msgId, msgData.emojiIndex)
+            MsgTypeEnum.LOCATION -> return LocationMsg.generate(msgData.msgId, msgData.text, msgData.lat, msgData.lng)
+            MsgTypeEnum.CUSTOM -> {
+                val data = msgData.data
+                val isBackLog = data.indexOf(Constant.BACKLOG_MSG) > 0
+                if (isBackLog) {
+                    msg.remark = Constant.BACKLOG_MSG
+                }
+                return CustomMsg.generate(msgData.msgId, data)
+            }
+
+            else -> {}
+        }
+        return null
+    }
+
+    private fun isTimeInRange(startTime: LocalTime, endTime: LocalTime): Boolean {
+        val currentTime = LocalTime.now()
+        if (endTime.isBefore(startTime)) {
+            return !currentTime.isBefore(startTime) || currentTime.isBefore(endTime)
+        }
+        return currentTime.isAfter(startTime) && currentTime.isBefore(endTime)
+    }
+
+    private fun MessageHistory.toMsg(): Message {
+        val msg = Message()
+        msg.msgId = this.id
+        msg.msgSummary = this.msgSummary
+        msg.sender = this.fromAccount
+        msg.receiver = this.toAccount
+        msg.sendTimestamp = this.sendTime.toInstant(ZoneOffset.of("+8")).toEpochMilli()
+        msg.msgTimestamp = this.createTime.time
+        return msg
+    }
+}
@@ -80,7 +80,7 @@ class MsgPrepareBizService(
         }
         val result = MessageV2Result()
         result.setMessageId(dto.id)
-        val msgState = BeanTransfer.trans2SdkMsgState(pair.left)
+        val msgState = BeanTransfer.trans2SdkMsgState(pair!!.left)
         result.setMsgState(msgState.value)
         result.setMsgStateDesc(pair.value)
         result.setResultTime(DateUtil.toDate(LocalDateTime.now()))
@@ -77,7 +77,7 @@ class XgPushService(
  
     fun sendMsg(userId: Long, accountType: AccountTypeEnum): Pair<Int, String> {
         val xgToken = getUserToken(userId, accountType)
-        if (Constant.inValidToken == xgToken) {
+        if (Constant.INVALID_TOKEN == xgToken) {
             return Pair(-1, "")
         }
         val pushAppRequest = PushAppRequest()
@@ -107,9 +107,9 @@ class XgPushService(
         return Pair(code, if (code == 0) pushId else errMsg)
     }
  
-    fun sendData(userId: Long, accountType: AccountTypeEnum) {
+    fun sendDebugData(userId: Long, accountType: AccountTypeEnum) {
         val xgToken = getUserToken(userId, accountType)
-        BV.isFalse(Constant.inValidToken == xgToken) {
+        BV.isFalse(Constant.INVALID_TOKEN == xgToken) {
             "用户设备未找到"
         }
         val pushAppRequest = PushAppRequest()
@@ -135,7 +135,7 @@ class XgPushService(
     private fun getUserToken(userId: Long, accountType: AccountTypeEnum): String {
         val cacheKey = "${keyPrefix}:${accountType.name}"
         val hashOps = stringRedisTemplate.boundHashOps<String, String>(cacheKey)
-        return hashOps.get(userId.toString()) ?: hermesService.queryByUserId(userId, accountType)?.xgToken ?: Constant.inValidToken
+        return hashOps.get(userId.toString()) ?: hermesService.queryByUserId(userId, accountType)?.xgToken ?: Constant.INVALID_TOKEN
     }
  
     private fun clearTokenAccount(token: String, account: String) {
@@ -166,7 +166,7 @@ class XgPushService(
         if (list.isNullOrEmpty()) {
             return listOf()
         }
-        list.forEach { it.xgToken = Constant.inValidToken }
+        list.forEach { it.xgToken = Constant.INVALID_TOKEN }
         hermesService.updateBatchById(list)
         return list.map(Hermes::getUserId)
     }
@@ -10,7 +10,6 @@ import cn.fw.hermes.domain.mqtt.UpdateMessage
 import cn.fw.hermes.service.data.MessageHistoryService
 import cn.fw.hermes.service.property.MqttProperty
 import cn.hutool.json.JSONUtil
-import lombok.extern.slf4j.Slf4j
 import org.apache.commons.lang3.tuple.MutablePair
 import org.apache.commons.lang3.tuple.Pair
 import org.eclipse.paho.client.mqttv3.*
@@ -28,7 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean
  * @desc : mqtt服务
  * @date : 2023-07-20 17:46
  */
-@Slf4j
 @Service
 class MqttService(
     private val mqttProperty: MqttProperty,