websocket消息逻辑整理

themis-business/src/main/java/com/qmth/themis/business/service/impl/TMRocketMessageServiceImpl.java

@@ -33,11 +33,7 @@ public class TMRocketMessageServiceImpl extends ServiceImpl<TMRocketMessageMappe
     @Override
     public void saveMqMessageSuccess(MqDto mqDto, String key) {
         mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-        Gson gson = new Gson();
-        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
-        this.saveOrUpdate(tmRocketMessage);
-        redisUtil.delete(key, mqDto.getId());
+        this.saveMessage(mqDto, key);
     }
 
     /**
@@ -49,6 +45,16 @@ public class TMRocketMessageServiceImpl extends ServiceImpl<TMRocketMessageMappe
     @Override
     public void saveMqMessageError(MqDto mqDto, String key) {
         mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
+        this.saveMessage(mqDto, key);
+    }
+
+    /**
+     * 保存消息
+     *
+     * @param mqDto
+     * @param key
+     */
+    protected void saveMessage(MqDto mqDto, String key) {
         Gson gson = new Gson();
         TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
         tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));