fix: 修复集群消息转换 bug

Author: Amazingwujun

readme.md

@@ -159,7 +159,7 @@
 ![ak6nHK.png](https://s1.ax1x.com/2020/07/28/ak6nHK.png)
 
 1. `mqttx.cluster.enable`：功能开关，默认 `false`
-
+> `v1.0.5.RELEASE` 之前的版本集群消息处理存在 bug，无法使用.
 #### 4.4 ssl 支持
 
 开启 ssl 你首先应该有了 *ca*(自签名或购买)，然后修改 `application.yml` 文件中几个配置：

src/main/java/com/jun/mqttx/broker/BrokerHandler.java

@@ -1,6 +1,7 @@
 package com.jun.mqttx.broker;
 
-import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
 import com.jun.mqttx.broker.handler.AbstractMqttSessionHandler;
 import com.jun.mqttx.broker.handler.ConnectHandler;
 import com.jun.mqttx.broker.handler.MessageDelegatingHandler;
@@ -44,7 +45,7 @@
 @Slf4j
 @ChannelHandler.Sharable
 @Component
-public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> implements Watcher<Object> {
+public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> implements Watcher {
 
     /**
      * channel 群组
@@ -234,26 +235,26 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
      *     <li>修改用户权限</li>
      * </ol>
      *
-     * @param im {@see InternalMessage}
+     * @param msg 集群消息
      */
     @Override
-    public void action(InternalMessage<Object> im) {
-        if (im.getData() instanceof JSONObject) {
-            Authentication data = ((JSONObject) im.getData()).toJavaObject(Authentication.class);
-            // 目的是为了兼容 v1.0.2(含) 之前的版本
-            String clientId = StringUtils.isEmpty(data.getClientId()) ? data.getUsername() : data.getClientId();
-            List<String> authorizedPub = data.getAuthorizedPub();
-            List<String> authorizedSub = data.getAuthorizedSub();
-            if (StringUtils.isEmpty(clientId) || (CollectionUtils.isEmpty(authorizedPub) && CollectionUtils.isEmpty(authorizedSub))) {
-                log.info("权限修改参数非法:{}", im);
-                return;
-            }
-            alterUserAuthorizedTopic(clientId, authorizedSub, authorizedPub);
+    public void action(String msg) {
+        InternalMessage<Authentication> im = JSON.parseObject(msg, new TypeReference<InternalMessage<Authentication>>() {
+        });
+        Authentication data = im.getData();
+        // 目的是为了兼容 v1.0.2(含) 之前的版本
+        String clientId = StringUtils.isEmpty(data.getClientId()) ? data.getUsername() : data.getClientId();
+        List<String> authorizedPub = data.getAuthorizedPub();
+        List<String> authorizedSub = data.getAuthorizedSub();
+        if (StringUtils.isEmpty(clientId) || (CollectionUtils.isEmpty(authorizedPub) && CollectionUtils.isEmpty(authorizedSub))) {
+            log.info("权限修改参数非法:{}", im);
+            return;
+        }
+        alterUserAuthorizedTopic(clientId, authorizedSub, authorizedPub);
 
-            // 移除 cache&redis 中客户端订阅的 topic
-            if (!CollectionUtils.isEmpty(authorizedSub)) {
-                subscriptionService.clearClientSub(clientId, authorizedSub);
-            }
+        // 移除 cache&redis 中客户端订阅的 topic
+        if (!CollectionUtils.isEmpty(authorizedSub)) {
+            subscriptionService.clearClientSub(clientId, authorizedSub);
         }
     }
 

src/main/java/com/jun/mqttx/broker/handler/DisconnectHandler.java

@@ -1,5 +1,7 @@
 package com.jun.mqttx.broker.handler;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
 import com.jun.mqttx.broker.BrokerHandler;
 import com.jun.mqttx.constants.InternalMessageEnum;
 import com.jun.mqttx.consumer.Watcher;
@@ -20,7 +22,7 @@
  */
 @Slf4j
 @Handler(type = MqttMessageType.DISCONNECT)
-public final class DisconnectHandler extends AbstractMqttSessionHandler implements Watcher<String> {
+public final class DisconnectHandler extends AbstractMqttSessionHandler implements Watcher {
 
     private ConnectHandler connectHandler;
 
@@ -46,7 +48,9 @@ public void process(ChannelHandlerContext ctx, MqttMessage msg) {
     }
 
     @Override
-    public void action(InternalMessage<String> im) {
+    public void action(String msg) {
+        InternalMessage<String> im = JSON.parseObject(msg, new TypeReference<InternalMessage<String>>() {
+        });
         ChannelId channelId = ConnectHandler.CLIENT_MAP.get(im.getData());
         if (channelId != null) {
             BrokerHandler.CHANNELS.find(channelId).close();

src/main/java/com/jun/mqttx/broker/handler/PublishHandler.java

@@ -1,5 +1,7 @@
 package com.jun.mqttx.broker.handler;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
 import com.jun.mqttx.broker.BrokerHandler;
 import com.jun.mqttx.config.MqttxConfig;
 import com.jun.mqttx.constants.InternalMessageEnum;
@@ -36,7 +38,7 @@
  * @since 1.0.4
  */
 @Handler(type = MqttMessageType.PUBLISH)
-public class PublishHandler extends AbstractMqttTopicSecureHandler implements Watcher<PubMsg> {
+public class PublishHandler extends AbstractMqttTopicSecureHandler implements Watcher {
 
     private IRetainMessageService retainMessageService;
 
@@ -65,6 +67,7 @@ public class PublishHandler extends AbstractMqttTopicSecureHandler implements Wa
      */
     private Map<String, AtomicInteger> roundMap;
 
+    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
     public PublishHandler(IPublishMessageService publishMessageService, IRetainMessageService retainMessageService,
                           ISubscriptionService subscriptionService, IPubRelMessageService pubRelMessageService,
                           @Nullable IInternalMessagePublishService internalMessagePublishService, MqttxConfig mqttxConfig,
@@ -182,7 +185,9 @@ public void process(ChannelHandlerContext ctx, MqttMessage msg) {
     }
 
     @Override
-    public void action(InternalMessage<PubMsg> im) {
+    public void action(String msg) {
+        InternalMessage<PubMsg> im = JSON.parseObject(msg, new TypeReference<InternalMessage<PubMsg>>() {
+        });
         PubMsg data = im.getData();
         publish(data, null, true);
     }

src/main/java/com/jun/mqttx/consumer/InternalMessageSubscriber.java

@@ -45,7 +45,6 @@ public InternalMessageSubscriber(List<Watcher> watchers, MqttxConfig mqttxConfig
      * @param message 消息内容
      * @param channel 订阅频道
      */
-    @SuppressWarnings("unchecked")
     public void handleMessage(String message, String channel) {
         // 同 broker 消息屏蔽
         InternalMessage internalMessage = JSON.parseObject(message, InternalMessage.class);
@@ -55,7 +54,7 @@ public void handleMessage(String message, String channel) {
 
         for (Watcher watcher : watchers) {
             if (watcher.support(channel)) {
-                watcher.action(JSON.parseObject(message, InternalMessage.class));
+                watcher.action(message);
                 // 一个消息只能由一个观察者消费
                 break;
             }

src/main/java/com/jun/mqttx/consumer/Watcher.java

@@ -1,23 +1,22 @@
 package com.jun.mqttx.consumer;
 
 import com.jun.mqttx.constants.InternalMessageEnum;
-import com.jun.mqttx.entity.InternalMessage;
 
 /**
  * 观察者，实现此接口.
  *
  * @author Jun
  * @since 1.0.4
  */
-public interface Watcher<T> {
+public interface Watcher {
 
     /**
      * 每当有新的集群消息达到是，触发行为。
      * 注意：实现方法不应该有耗时操作(e.g. 访问数据库)
      *
-     * @param im {@see InternalMessage}
+     * @param msg 集群消息
      */
-    void action(InternalMessage<T> im);
+    void action(String msg);
 
     /**
      * Watcher 支持的 channel 类别

src/main/java/com/jun/mqttx/service/impl/SubscriptionServiceImpl.java

@@ -1,5 +1,7 @@
 package com.jun.mqttx.service.impl;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
 import com.jun.mqttx.config.MqttxConfig;
 import com.jun.mqttx.constants.InternalMessageEnum;
 import com.jun.mqttx.consumer.Watcher;
@@ -28,7 +30,7 @@
  */
 @Slf4j
 @Service
-public class SubscriptionServiceImpl implements ISubscriptionService, Watcher<ClientSubOrUnsubMsg> {
+public class SubscriptionServiceImpl implements ISubscriptionService, Watcher {
 
     /**
      * 按顺序 -> 订阅，解除订阅，删除 topic
@@ -229,8 +231,10 @@ public void clearClientSub(String clientId, List<String> authorizedSub) {
     }
 
     @Override
-    public void action(InternalMessage<ClientSubOrUnsubMsg> im) {
+    public void action(String msg) {
         if (enableInnerCache) {
+            InternalMessage<ClientSubOrUnsubMsg> im = JSON.parseObject(msg, new TypeReference<InternalMessage<ClientSubOrUnsubMsg>>() {
+            });
             ClientSubOrUnsubMsg data = im.getData();
             int type = data.getType();
             switch (type) {