新增判断

smqtt-bootstrap/src/test/java/ClusterNode1.java

@@ -9,7 +9,7 @@ public class ClusterNode1 {
 
     public static void main(String[] args) throws InterruptedException {
         Bootstrap bootstrap = Bootstrap.builder()
-                .rootLevel(Level.INFO)
+                .rootLevel(Level.ERROR)
                 .websocketConfig(
                         BootstrapConfig.WebsocketConfig
                                 .builder()

smqtt-bootstrap/src/test/java/topic/TopicTest.java

+package topic;
+
+import io.github.quickmsg.common.channel.MqttChannel;
+import io.github.quickmsg.common.topic.SubscribeTopic;
+import io.github.quickmsg.common.topic.TopicRegistry;
+import io.github.quickmsg.core.spi.DefaultTopicRegistry;
+import io.netty.handler.codec.mqtt.MqttQoS;
+
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.*;
+
+/**
+ * @author luxurong
+ */
+public class TopicTest {
+
+   static ExecutorService service = Executors.newFixedThreadPool(100);
+
+    private static TopicRegistry topicRegistry = new DefaultTopicRegistry();
+
+    private static Map<Integer, MqttChannel> channelMap  = new ConcurrentHashMap<>();
+
+    public static void main(String[] args) {
+        CountDownLatch count = new CountDownLatch(500000);
+        for(int i=0;i<500000;i++){
+            service.execute(()->{
+                int index = new Random().nextInt(1000);
+                MqttChannel mqttChannel =channelMap.computeIfAbsent(index,in->{
+                    MqttChannel mqttChannel1=new MqttChannel();
+                    mqttChannel1.setTopics(new CopyOnWriteArraySet<>());
+                    return mqttChannel1;
+                });
+                SubscribeTopic subscribeTopic=new SubscribeTopic(String.valueOf(index), MqttQoS.AT_MOST_ONCE,mqttChannel);
+                topicRegistry.registrySubscribeTopic(subscribeTopic);
+                topicRegistry.getAllTopics();
+                count.countDown();
+            });
+        }
+        try {
+            count.await();
+            Map<String, Set<MqttChannel>> topics =  topicRegistry.getAllTopics();
+            System.out.println(topics);
+
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+    }
+
+
+}

smqtt-common/src/main/java/io/github/quickmsg/common/channel/MqttChannel.java

@@ -188,7 +188,7 @@ public class MqttChannel {
      */
     public Mono<Void> write(MqttMessage mqttMessage, boolean retry) {
         // http本地mock
-        if (this.getIsMock()) {
+        if (this.getIsMock() && !this.active()) {
             return Mono.empty();
         } else {
             return MqttMessageSink.MQTT_SINK.sendMessage(mqttMessage, this, retry, replyMqttMessageMap);
@@ -346,7 +346,7 @@ public class MqttChannel {
     @Override
     public String toString() {
         return "MqttChannel{" +
-                " address='" + this.connection.address().toString() + '\'' +
+//                " address='" + this.connection.address().toString() + '\'' +
                 ", clientIdentifier='" + clientIdentifier + '\'' +
                 ", status=" + status +
                 ", keepalive=" + keepalive +

smqtt-common/src/main/java/io/github/quickmsg/common/enums/Event.java

@@ -79,8 +79,8 @@ public enum Event {
     /**
      * body
      *
-     * @param mqttChannel    {@link MqttChannel }
-     * @param body           {@link Object }
+     * @param mqttChannel {@link MqttChannel }
+     * @param body        {@link Object }
      * @return ByteBuf
      */
     public abstract ByteBuf writeBody(MqttChannel mqttChannel, Object body);
@@ -92,6 +92,9 @@ public enum Event {
                                 message
                                 , System.currentTimeMillis(), Boolean.FALSE),
                         receiveContext);
+        if (message instanceof MqttPublishMessage) {
+            ((MqttPublishMessage) message).release();
+        }
     }
 
 }

smqtt-core/src/main/java/io/github/quickmsg/core/spi/DefaultProtocolAdaptor.java

@@ -45,11 +45,11 @@ public class DefaultProtocolAdaptor implements ProtocolAdaptor {
     @Override
     public <C extends Configuration> void chooseProtocol(MqttChannel mqttChannel, SmqttMessage<MqttMessage> smqttMessage, ReceiveContext<C> receiveContext) {
         MqttMessage mqttMessage = smqttMessage.getMessage();
-        log.info(" 【{}】【{}】 【{}】",
-                Thread.currentThread().getName(),
-                mqttMessage.fixedHeader().messageType(),
-                mqttChannel);
         if (mqttMessage.decoderResult() != null && (mqttMessage.decoderResult().isSuccess())) {
+            log.info(" 【{}】【{}】 【{}】",
+                    Thread.currentThread().getName(),
+                    mqttMessage.fixedHeader().messageType(),
+                    mqttChannel);
             Optional.ofNullable(types.get(mqttMessage.fixedHeader().messageType()))
                     .ifPresent(protocol -> protocol
                             .doParseProtocol(smqttMessage, mqttChannel)