[broker-23] add splitTopic method, rework subscribeAck handler

Author: crazyrokr

src/main/java/com/ss/mqtt/broker/handler/packet/in/SubscribeInPacketHandler.java

@@ -2,6 +2,7 @@
 
 import static com.ss.mqtt.broker.model.reason.code.SubscribeAckReasonCode.SHARED_SUBSCRIPTIONS_NOT_SUPPORTED;
 import static com.ss.mqtt.broker.model.reason.code.SubscribeAckReasonCode.WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED;
+import static java.lang.Byte.toUnsignedInt;
 import com.ss.mqtt.broker.model.reason.code.DisconnectReasonCode;
 import com.ss.mqtt.broker.model.reason.code.SubscribeAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient.UnsafeMqttClient;
@@ -10,26 +11,27 @@
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.Set;
+
 @RequiredArgsConstructor
 public class SubscribeInPacketHandler extends AbstractPacketHandler<UnsafeMqttClient, SubscribeInPacket> {
 
+    private final static Set<SubscribeAckReasonCode> INVALID_ACK_CODE = Set.of(
+        SHARED_SUBSCRIPTIONS_NOT_SUPPORTED,
+        WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED
+    );
+
     private final @NotNull SubscriptionService subscriptionService;
 
     @Override
     protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull SubscribeInPacket packet) {
         var ackReasonCodes = subscriptionService.subscribe(client, packet.getTopicFilters());
         client.send(client.getPacketOutFactory().newSubscribeAck(packet.getPacketId(), ackReasonCodes));
-        ackReasonCodes.findAny(client, SubscribeInPacketHandler::hasInvalidReason);
-    }
-
-    private static boolean hasInvalidReason(@NotNull UnsafeMqttClient client, @NotNull SubscribeAckReasonCode reason) {
-        if (reason == SHARED_SUBSCRIPTIONS_NOT_SUPPORTED || reason == WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED) {
-            int reasonIndex = Byte.toUnsignedInt(reason.getValue());
-            var disconnect = client.getPacketOutFactory().newDisconnect(client, DisconnectReasonCode.of(reasonIndex));
+        var reason = ackReasonCodes.findAny(INVALID_ACK_CODE::contains);
+        if (reason != null) {
+            var disconnectReasonCode = DisconnectReasonCode.of(toUnsignedInt(reason.getValue()));
+            var disconnect = client.getPacketOutFactory().newDisconnect(client, disconnectReasonCode);
             client.sendWithFeedback(disconnect).thenAccept(result -> client.getConnection().close());
-            return true;
-        } else {
-            return false;
         }
     }
 }

src/main/java/com/ss/mqtt/broker/model/SingleSubscriber.java

@@ -7,7 +7,7 @@
 import org.jetbrains.annotations.NotNull;
 
 @ToString
-@EqualsAndHashCode(of = "mqttClient", callSuper = false)
+@EqualsAndHashCode(of = "mqttClient")
 public class SingleSubscriber implements Subscriber {
 
     private final @Getter @NotNull MqttClient mqttClient;

src/main/java/com/ss/mqtt/broker/model/topic/AbstractTopic.java

@@ -1,13 +1,14 @@
 package com.ss.mqtt.broker.model.topic;
 
-import static com.ss.mqtt.broker.util.TopicUtils.DELIMITER;
-import static com.ss.rlib.common.util.ArrayUtils.EMPTY_STRING_ARRAY;
+import static com.ss.mqtt.broker.util.TopicUtils.splitTopic;
 import static com.ss.rlib.common.util.StringUtils.EMPTY;
 import com.ss.mqtt.broker.util.DebugUtils;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.List;
+
 @Getter
 @EqualsAndHashCode(of = "rawTopic")
 public abstract class AbstractTopic {
@@ -16,28 +17,28 @@ public abstract class AbstractTopic {
         DebugUtils.registerIncludedFields("rawTopic");
     }
 
-    private final @NotNull String[] segments;
+    private final @NotNull List<String> segments;
     private final @NotNull String rawTopic;
     private final int length;
 
     AbstractTopic() {
         length = 0;
-        segments = EMPTY_STRING_ARRAY;
+        segments = List.of();
         rawTopic = EMPTY;
     }
 
     AbstractTopic(@NotNull String topicName) {
         length = topicName.length();
-        segments = topicName.split(DELIMITER);
+        segments = splitTopic(topicName);
         rawTopic = topicName;
     }
 
     @NotNull String getSegment(int level) {
-        return segments[level];
+        return segments.get(level);
     }
 
     int levelsCount() {
-        return segments.length;
+        return segments.size();
     }
 
     @Override

src/main/java/com/ss/mqtt/broker/util/TopicUtils.java

@@ -1,18 +1,22 @@
 package com.ss.mqtt.broker.util;
 
+import static java.util.Collections.unmodifiableList;
 import com.ss.mqtt.broker.model.topic.SharedTopicFilter;
 import com.ss.mqtt.broker.model.topic.TopicFilter;
 import com.ss.mqtt.broker.model.topic.TopicName;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.ArrayList;
+import java.util.List;
+
 public class TopicUtils {
 
     private static final TopicFilter INVALID_TOPIC_FILTER = new TopicFilter();
     private static final TopicName INVALID_TOPIC_NAME = new TopicName();
     public static final TopicName EMPTY_TOPIC_NAME = new TopicName();
 
     private static final String SHARE_KEYWORD = "$share";
-    public static final String DELIMITER = "/";
+    private static final String DELIMITER = "/";
     public static final String MULTI_LEVEL_WILDCARD = "#";
     public static final String SINGLE_LEVEL_WILDCARD = "+";
 
@@ -51,6 +55,17 @@ public static boolean hasWildcard(@NotNull TopicFilter topicFilter) {
         }
     }
 
+    public static @NotNull List<String> splitTopic(@NotNull String topic) {
+        var segments = new ArrayList<String>();
+        int pos = 0, end;
+        while ((end = topic.indexOf(DELIMITER, pos)) >= 0) {
+            segments.add(topic.substring(pos, end));
+            pos = end + 1;
+        }
+        segments.add(topic.substring(pos));
+        return unmodifiableList(segments);
+    }
+
     private static @NotNull TopicFilter newSharedTopicFilter(@NotNull String topicFilter) {
         int firstSlash = topicFilter.indexOf(DELIMITER) + 1;
         int secondSlash = topicFilter.indexOf(DELIMITER, firstSlash);