[broker-23] refactor TopicSubscribers

Author: crazyrokr

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

@@ -6,6 +6,7 @@
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
 import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.network.client.MqttClient;
+import com.ss.mqtt.broker.util.SubscriberUtils;
 import com.ss.rlib.common.function.NotNullSupplier;
 import com.ss.rlib.common.util.array.Array;
 import com.ss.rlib.common.util.array.ConcurrentArray;
@@ -19,14 +20,6 @@ public class TopicSubscribers {
 
     private final static NotNullSupplier<TopicSubscribers> TOPIC_SUBSCRIBER_SUPPLIER = TopicSubscribers::new;
 
-    private static boolean isSharedSubscriber(@NotNull Subscriber subscriber){
-        return subscriber instanceof SharedSubscriber;
-    }
-
-    private static boolean isSingleSubscriber(@NotNull Subscriber subscriber){
-        return subscriber instanceof SingleSubscriber;
-    }
-
     private static void addSubscriber(
         @NotNull ConcurrentArray<Subscriber> subscribers,
         @NotNull MqttClient client,
@@ -56,7 +49,7 @@ private static void addSharedSubscriber(
 
         var sharedSubscriber = (SharedSubscriber) subscribers.findAny(
             group,
-            TopicSubscribers::sharedSubscriberWithGroup
+            SubscriberUtils::sharedSubscriberWithGroup
         );
 
         if (sharedSubscriber == null) {
@@ -68,14 +61,6 @@ private static void addSharedSubscriber(
         sharedSubscriber.addSubscriber(singleSubscriber);
     }
 
-    private static boolean sharedSubscriberWithGroup(@NotNull String group, @NotNull Subscriber subscriber) {
-        return isSharedSubscriber(subscriber) && group.equals(((SharedSubscriber) subscriber).getGroup());
-    }
-
-    private static @Nullable MqttClient singleSubscriberToMqttClient(@NotNull Subscriber subscriber) {
-        return isSingleSubscriber(subscriber) ? ((SingleSubscriber) subscriber).getMqttClient() : null;
-    }
-
     private static boolean removeSubscriber(
         @NotNull ConcurrentArray<Subscriber> subscribers,
         @NotNull TopicFilter topic,
@@ -93,7 +78,7 @@ private static boolean removeSingleSubscriber(
         //noinspection NullableProblems
         return subscribers.removeIfConverted(
             client,
-            TopicSubscribers::singleSubscriberToMqttClient,
+            SubscriberUtils::singleSubscriberToMqttClient,
             Object::equals
         );
     }
@@ -106,7 +91,7 @@ private static boolean removeSharedSubscriber(
         boolean removed = false;
         var sharedSubscriber = (SharedSubscriber) subscribers.findAny(
             group,
-            TopicSubscribers::sharedSubscriberWithGroup
+            SubscriberUtils::sharedSubscriberWithGroup
         );
 
         if (sharedSubscriber != null) {
@@ -138,7 +123,7 @@ private static boolean removeDuplicateWithLowerQoS(
         }
     }
 
-    private static void addSubscriber(@NotNull Array<SingleSubscriber> result, @NotNull Subscriber subscriber) {
+    private static void addToResultArray(@NotNull Array<SingleSubscriber> result, @NotNull Subscriber subscriber) {
         if (subscriber instanceof SharedSubscriber) {
             result.add(((SharedSubscriber) subscriber).getSubscriber());
         } else {
@@ -163,7 +148,7 @@ private static void addSubscriber(@NotNull Array<SingleSubscriber> result, @NotN
                 subscribers.forEachFiltered(
                     result,
                     TopicSubscribers::removeDuplicateWithLowerQoS,
-                    TopicSubscribers::addSubscriber
+                    TopicSubscribers::addToResultArray
                 );
             } finally {
                 subscribers.readUnlock(stamp);
@@ -176,10 +161,10 @@ private static void addSubscriber(@NotNull Array<SingleSubscriber> result, @NotN
     private volatile @Getter @Nullable ConcurrentArray<Subscriber> subscribers;
 
     public void addSubscriber(@NotNull MqttClient client, @NotNull SubscribeTopicFilter subscribe) {
-        addSubscriber(0, subscribe.getTopicFilter(), client, subscribe);
+        searchPlaceForSubscriber(0, subscribe.getTopicFilter(), client, subscribe);
     }
 
-    private void addSubscriber(
+    private void searchPlaceForSubscriber(
         int level,
         @NotNull TopicFilter topicFilter,
         @NotNull MqttClient client,
@@ -198,7 +183,7 @@ private void addSubscriber(
                 ObjectDictionary::getOrCompute
             );
             //noinspection ConstantConditions
-            topicSubscriber.addSubscriber(level + 1, topicFilter, client, subscribe);
+            topicSubscriber.searchPlaceForSubscriber(level + 1, topicFilter, client, subscribe);
         }
     }
 
@@ -207,27 +192,27 @@ public void removeSubscriber(@NotNull MqttClient client, @NotNull SubscribeTopic
     }
 
     public boolean removeSubscriber(@NotNull MqttClient client, @NotNull TopicFilter topicFilter) {
-        return removeSubscriber(0, topicFilter, client);
+        return searchSubscriberToRemove(0, topicFilter, client);
     }
 
-    private boolean removeSubscriber(int level, @NotNull TopicFilter topicFilter, @NotNull MqttClient mqttClient) {
+    private boolean searchSubscriberToRemove(int level, @NotNull TopicFilter topicFilter, @NotNull MqttClient mqttClient) {
         var removed = false;
         var topicSubscribers = getTopicSubscribers();
         if (level == topicFilter.levelsCount()) {
-            removed = removeSubscriber(topicFilter, mqttClient);
+            removed = tryToRemoveSubscriber(topicFilter, mqttClient);
         } else if (topicSubscribers != null) {
             var topicSubscriber = topicSubscribers.getInReadLock(
                 topicFilter.getSegment(level),
                 ObjectDictionary::get
             );
             if (topicSubscriber != null) {
-                removed = topicSubscriber.removeSubscriber(level + 1, topicFilter, mqttClient);
+                removed = topicSubscriber.searchSubscriberToRemove(level + 1, topicFilter, mqttClient);
             }
         }
         return removed;
     }
 
-    private boolean removeSubscriber(@NotNull TopicFilter topicFilter, @NotNull MqttClient mqttClient) {
+    private boolean tryToRemoveSubscriber(@NotNull TopicFilter topicFilter, @NotNull MqttClient mqttClient) {
         var removed = false;
         var subscribers = getSubscribers();
         if (subscribers != null) {

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

@@ -0,0 +1,27 @@
+package com.ss.mqtt.broker.util;
+
+import com.ss.mqtt.broker.model.SharedSubscriber;
+import com.ss.mqtt.broker.model.SingleSubscriber;
+import com.ss.mqtt.broker.model.Subscriber;
+import com.ss.mqtt.broker.network.client.MqttClient;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public class SubscriberUtils {
+
+    private static boolean isSharedSubscriber(@NotNull Subscriber subscriber) {
+        return subscriber instanceof SharedSubscriber;
+    }
+
+    private static boolean isSingleSubscriber(@NotNull Subscriber subscriber) {
+        return subscriber instanceof SingleSubscriber;
+    }
+
+    public static boolean sharedSubscriberWithGroup(@NotNull String group, @NotNull Subscriber subscriber) {
+        return isSharedSubscriber(subscriber) && group.equals(((SharedSubscriber) subscriber).getGroup());
+    }
+
+    public static @Nullable MqttClient singleSubscriberToMqttClient(@NotNull Subscriber subscriber) {
+        return isSingleSubscriber(subscriber) ? ((SingleSubscriber) subscriber).getMqttClient() : null;
+    }
+}