feat: 系统主题功能支持

1. 一些小改动

Author: Amazingwujun

pom.xml

@@ -22,7 +22,7 @@
     <dependencies>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
         </dependency>
 
         <dependency>

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

@@ -10,12 +10,10 @@
 import com.jun.mqttx.utils.TopicUtils;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.handler.codec.mqtt.*;
-import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.GlobalEventExecutor;
 
 import java.util.ArrayList;
@@ -34,9 +32,6 @@
 @Handler(type = MqttMessageType.SUBSCRIBE)
 public class SubscribeHandler extends AbstractMqttTopicSecureHandler {
     //@formatter:off
-    /** 客户端订阅的系统主题 key, 用于 {@link Channel#attr( AttributeKey)} */
-    public static final String SYS_TOPICS_ATTR = "sys_topics";
-
     private final boolean enableTopicPubSubSecure;
     private IRetainMessageService retainMessageService;
     private ISubscriptionService subscriptionService;
@@ -46,9 +41,13 @@ public class SubscribeHandler extends AbstractMqttTopicSecureHandler {
     private MqttQoS sysTopicQos;
     private String version;
 
-    /** 系统主题 $SYS 订阅群组 */
-    public static final ChannelGroup sysChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
-    /** 定时任务执行器 */
+    /**
+     * 系统主题 $SYS 订阅群组
+     */
+    public static ChannelGroup sysChannels;
+    /**
+     * 定时任务执行器
+     */
     private ScheduledExecutorService fixRateExecutor;
     /** 系统主题消息 id  */
     private AtomicInteger sysTopicMsgId;
@@ -64,6 +63,7 @@ public SubscribeHandler(IRetainMessageService retainMessageService, ISubscriptio
         this.version = mqttxConfig.getVersion();
         this.enableSysTopic = mqttxConfig.getSysTopic().getEnable();
         if (enableSysTopic) {
+            SubscribeHandler.sysChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
             this.interval = mqttxConfig.getSysTopic().getInterval().getSeconds();
             this.sysTopicQos = MqttQoS.valueOf(mqttxConfig.getSysTopic().getQos());
             fixRateExecutor = Executors.newSingleThreadScheduledExecutor();

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

@@ -1,5 +1,6 @@
 package com.jun.mqttx.broker.handler;
 
+import com.jun.mqttx.config.MqttxConfig;
 import com.jun.mqttx.service.ISubscriptionService;
 import com.jun.mqttx.utils.TopicUtils;
 import io.netty.channel.Channel;
@@ -18,9 +19,12 @@
 @Handler(type = MqttMessageType.UNSUBSCRIBE)
 public class UnsubscribeHandler extends AbstractMqttSessionHandler {
 
+    private Boolean enableSysTopic;
+
     private ISubscriptionService subscriptionService;
 
-    public UnsubscribeHandler(ISubscriptionService subscriptionService) {
+    public UnsubscribeHandler(MqttxConfig config, ISubscriptionService subscriptionService) {
+        this.enableSysTopic = config.getSysTopic().getEnable();
         this.subscriptionService = subscriptionService;
     }
 
@@ -31,7 +35,10 @@ public void process(ChannelHandlerContext ctx, MqttMessage msg) {
         MqttUnsubscribePayload payload = mqttUnsubscribeMessage.payload();
 
         // 系统主题
-        List<String> collect = unsubscribeSysTopics(payload.topics(), ctx.channel());
+        List<String> collect = payload.topics();
+        if (enableSysTopic) {
+            collect = unsubscribeSysTopics(payload.topics(), ctx.channel());
+        }
 
         // 非系统主题
         subscriptionService.unsubscribe(clientId(ctx), collect);