[broker-14] work on code review

crazyrokr · crazyrokr · commit 7bfbd703e89d · 2019-11-28T21:09:39.000+03:00
diff --git a/build.gradle b/build.gradle
@@ -94,11 +94,7 @@ allprojects {
     }
     
     tasks.withType(GroovyCompile) {
-        options.forkOptions.jvmArgs << "--enable-preview"
-    }
-    
-    tasks.withType(JavaCompile) {
-        options.compilerArgs.add("--enable-preview")
+        options.forkOptions.jvmArgs += "--enable-preview"
     }
     
     processResources {
diff --git a/src/main/java/com/ss/mqtt/broker/config/MqttBrokerConfig.java b/src/main/java/com/ss/mqtt/broker/config/MqttBrokerConfig.java
@@ -13,7 +13,6 @@
 import com.ss.mqtt.broker.handler.publish.out.Qos2PublishOutHandler;
 import com.ss.mqtt.broker.model.MqttPropertyConstants;
 import com.ss.mqtt.broker.model.QoS;
-import com.ss.mqtt.broker.model.topic.TopicSubscribers;
 import com.ss.mqtt.broker.network.MqttConnection;
 import com.ss.mqtt.broker.network.client.DeviceMqttClient;
 import com.ss.mqtt.broker.network.client.MqttClient.UnsafeMqttClient;
@@ -69,11 +68,6 @@ private interface ChannelFactory extends
         );
     }
 
-    @Bean
-    @NotNull TopicSubscribers topicSubscribers() {
-        return new TopicSubscribers();
-    }
-
     @Bean
     @NotNull MqttSessionService mqttSessionService() {
         return new InMemoryMqttSessionService(
@@ -101,8 +95,7 @@ private interface ChannelFactory extends
         @NotNull SubscriptionService subscriptionService,
         @NotNull PublishingService publishingService,
         @NotNull MqttSessionService mqttSessionService,
-        @NotNull PublishRetryService publishRetryService,
-        @NotNull TopicSubscribers topicSubscribers
+        @NotNull PublishRetryService publishRetryService
     ) {
 
         var handlers = new PacketInHandler[PacketType.INVALID.ordinal()];
@@ -111,7 +104,7 @@ private interface ChannelFactory extends
             authenticationService,
             mqttSessionService,
             publishRetryService,
-            topicSubscribers
+            subscriptionService
         );
         handlers[PacketType.SUBSCRIBE.ordinal()] = new SubscribeInPacketHandler(subscriptionService);
         handlers[PacketType.UNSUBSCRIBE.ordinal()] = new UnsubscribeInPacketHandler(subscriptionService);
@@ -127,13 +120,13 @@ private interface ChannelFactory extends
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull MqttSessionService mqttSessionService,
         @NotNull PublishRetryService publishRetryService,
-        @NotNull TopicSubscribers topicSubscribers
+        @NotNull SubscriptionService subscriptionService
     ) {
         return new DefaultMqttClientReleaseHandler(
             clientIdRegistry,
             mqttSessionService,
             publishRetryService,
-            topicSubscribers
+            subscriptionService
         );
     }
 
@@ -179,8 +172,8 @@ private interface ChannelFactory extends
     }
 
     @Bean
-    @NotNull SubscriptionService subscriptionService(@NotNull TopicSubscribers topicSubscribers) {
-        return new SimpleSubscriptionService(topicSubscribers);
+    @NotNull SubscriptionService subscriptionService() {
+        return new SimpleSubscriptionService();
     }
 
     @Bean
diff --git a/src/main/java/com/ss/mqtt/broker/handler/client/AbstractMqttClientReleaseHandler.java b/src/main/java/com/ss/mqtt/broker/handler/client/AbstractMqttClientReleaseHandler.java
@@ -1,11 +1,11 @@
 package com.ss.mqtt.broker.handler.client;
 
-import com.ss.mqtt.broker.model.topic.TopicSubscribers;
 import com.ss.mqtt.broker.network.client.AbstractMqttClient;
 import com.ss.mqtt.broker.network.client.MqttClient.UnsafeMqttClient;
 import com.ss.mqtt.broker.service.ClientIdRegistry;
 import com.ss.mqtt.broker.service.MqttSessionService;
 import com.ss.mqtt.broker.service.PublishRetryService;
+import com.ss.mqtt.broker.service.SubscriptionService;
 import com.ss.rlib.common.util.StringUtils;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -20,7 +20,7 @@ public abstract class AbstractMqttClientReleaseHandler<T extends AbstractMqttCli
     private final @NotNull ClientIdRegistry clientIdRegistry;
     private final @NotNull MqttSessionService sessionService;
     private final @NotNull PublishRetryService publishRetryService;
-    private final @NotNull TopicSubscribers topicSubscribers;
+    private final @NotNull SubscriptionService subscriptionService;
 
     @Override
     public @NotNull Mono<?> release(@NotNull UnsafeMqttClient client) {
@@ -44,7 +44,7 @@ public abstract class AbstractMqttClientReleaseHandler<T extends AbstractMqttCli
         Mono<?> asyncActions = null;
 
         if (session != null) {
-            topicSubscribers.cleanSubscribers(client, session.getTopicFilters());
+            subscriptionService.cleanSubscriptions(client, session);
             if (client.getConnectionConfig().isSessionsEnabled()) {
                 asyncActions = sessionService.store(clientId, session, client.getSessionExpiryInterval());
                 client.setSession(null);
diff --git a/src/main/java/com/ss/mqtt/broker/handler/client/DefaultMqttClientReleaseHandler.java b/src/main/java/com/ss/mqtt/broker/handler/client/DefaultMqttClientReleaseHandler.java
@@ -1,10 +1,10 @@
 package com.ss.mqtt.broker.handler.client;
 
-import com.ss.mqtt.broker.model.topic.TopicSubscribers;
 import com.ss.mqtt.broker.network.client.DeviceMqttClient;
 import com.ss.mqtt.broker.service.ClientIdRegistry;
 import com.ss.mqtt.broker.service.MqttSessionService;
 import com.ss.mqtt.broker.service.PublishRetryService;
+import com.ss.mqtt.broker.service.SubscriptionService;
 import org.jetbrains.annotations.NotNull;
 
 public class DefaultMqttClientReleaseHandler extends AbstractMqttClientReleaseHandler<DeviceMqttClient> {
@@ -13,8 +13,8 @@ public DefaultMqttClientReleaseHandler(
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull MqttSessionService sessionService,
         @NotNull PublishRetryService publishRetryService,
-        @NotNull TopicSubscribers topicSubscribers
+        @NotNull SubscriptionService subscriptionService
     ) {
-        super(clientIdRegistry, sessionService, publishRetryService, topicSubscribers);
+        super(clientIdRegistry, sessionService, publishRetryService, subscriptionService);
     }
 }
diff --git a/src/main/java/com/ss/mqtt/broker/handler/packet/in/ConnectInPacketHandler.java b/src/main/java/com/ss/mqtt/broker/handler/packet/in/ConnectInPacketHandler.java
@@ -1,14 +1,13 @@
 package com.ss.mqtt.broker.handler.packet.in;
 
+import static com.ss.mqtt.broker.model.MqttPropertyConstants.*;
 import static com.ss.mqtt.broker.model.reason.code.ConnectAckReasonCode.BAD_USER_NAME_OR_PASSWORD;
 import static com.ss.mqtt.broker.model.reason.code.ConnectAckReasonCode.CLIENT_IDENTIFIER_NOT_VALID;
-import static com.ss.mqtt.broker.model.MqttPropertyConstants.*;
 import static com.ss.mqtt.broker.util.ReactorUtils.ifTrue;
 import com.ss.mqtt.broker.exception.ConnectionRejectException;
 import com.ss.mqtt.broker.exception.MalformedPacketMqttException;
-import com.ss.mqtt.broker.model.reason.code.ConnectAckReasonCode;
 import com.ss.mqtt.broker.model.MqttSession;
-import com.ss.mqtt.broker.model.topic.TopicSubscribers;
+import com.ss.mqtt.broker.model.reason.code.ConnectAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient.UnsafeMqttClient;
 import com.ss.mqtt.broker.network.packet.in.ConnectInPacket;
 import com.ss.mqtt.broker.service.*;
@@ -24,7 +23,7 @@ public class ConnectInPacketHandler extends AbstractPacketHandler<UnsafeMqttClie
     private final @NotNull AuthenticationService authenticationService;
     private final @NotNull MqttSessionService mqttSessionService;
     private final @NotNull PublishRetryService publishRetryService;
-    private final @NotNull TopicSubscribers topicSubscribers;
+    private final @NotNull SubscriptionService subscriptionService;
 
     @Override
     protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull ConnectInPacket packet) {
@@ -138,7 +137,7 @@ private Mono<Boolean> onConnected(
 
         publishRetryService.register(client);
 
-        topicSubscribers.restoreSubscribers(client, session.getTopicFilters());
+        subscriptionService.restoreSubscriptions(client, session);
 
         return Mono.just(Boolean.TRUE);
     }
diff --git a/src/main/java/com/ss/mqtt/broker/handler/publish/in/AbstractPublishInHandler.java b/src/main/java/com/ss/mqtt/broker/handler/publish/in/AbstractPublishInHandler.java
@@ -22,7 +22,7 @@ public void handle(@NotNull MqttClient client, @NotNull PublishInPacket packet)
             packet,
             this::publish
         );
-        handleImpl(client, packet, result);
+        handleResult(client, packet, result);
     }
 
     private boolean publish(
@@ -36,7 +36,7 @@ private boolean publish(
         return publishOutHandlers[qos.ordinal()];
     }
 
-    protected void handleImpl(
+    protected void handleResult(
         @NotNull MqttClient client,
         @NotNull PublishInPacket packet,
         @NotNull ActionResult result
diff --git a/src/main/java/com/ss/mqtt/broker/handler/publish/in/Qos1PublishInHandler.java b/src/main/java/com/ss/mqtt/broker/handler/publish/in/Qos1PublishInHandler.java
@@ -18,12 +18,12 @@ public Qos1PublishInHandler(
     }
 
     @Override
-    protected void handleImpl(
+    protected void handleResult(
         @NotNull MqttClient client,
         @NotNull PublishInPacket packet,
         @NotNull ActionResult result
     ) {
-        PublishAckReasonCode reasonCode = switch (result) {
+        var reasonCode = switch (result) {
             case EMPTY -> PublishAckReasonCode.NO_MATCHING_SUBSCRIBERS;
             case SUCCESS -> PublishAckReasonCode.SUCCESS;
             default -> PublishAckReasonCode.UNSPECIFIED_ERROR;
diff --git a/src/main/java/com/ss/mqtt/broker/model/MqttSession.java b/src/main/java/com/ss/mqtt/broker/model/MqttSession.java
@@ -1,9 +1,10 @@
 package com.ss.mqtt.broker.model;
 
+import com.ss.mqtt.broker.model.topic.TopicFilter;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.network.packet.HasPacketId;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
-import com.ss.rlib.common.util.array.ConcurrentArray;
+import com.ss.rlib.common.function.NotNullTripleConsumer;
 import org.jetbrains.annotations.NotNull;
 
 public interface MqttSession {
@@ -41,5 +42,12 @@ interface PendingPacketHandler {
 
     void registerPendingPublish(@NotNull PublishInPacket publish, @NotNull PendingPacketHandler handler, int packetId);
     void updatePendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);
-    @NotNull ConcurrentArray<SubscribeTopicFilter> getTopicFilters();
+
+    <F, S> void forEachTopicFilter(
+        @NotNull F first,
+        @NotNull S second,
+        @NotNull NotNullTripleConsumer<F, S, SubscribeTopicFilter> consumer
+    );
+    void addSubscriber(@NotNull SubscribeTopicFilter subscribe);
+    void removeSubscriber(@NotNull TopicFilter subscribe);
 }
diff --git a/src/main/java/com/ss/mqtt/broker/model/Subscriber.java b/src/main/java/com/ss/mqtt/broker/model/Subscriber.java
@@ -25,11 +25,11 @@ public Subscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter
         this.subscribeTopicFilter = topicFilter;
     }
 
-    public QoS getQos() {
+    public @NotNull QoS getQos() {
         return subscribeTopicFilter.getQos();
     }
 
-    public TopicFilter getTopicFilter() {
+    public @NotNull TopicFilter getTopicFilter() {
         return subscribeTopicFilter.getTopicFilter();
     }
 
diff --git a/src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java b/src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java
@@ -3,9 +3,11 @@
 import com.ss.mqtt.broker.model.MqttPropertyConstants;
 import com.ss.mqtt.broker.model.MqttSession.UnsafeMqttSession;
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
+import com.ss.mqtt.broker.model.topic.TopicFilter;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.network.packet.HasPacketId;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
+import com.ss.rlib.common.function.NotNullTripleConsumer;
 import com.ss.rlib.common.util.ClassUtils;
 import com.ss.rlib.common.util.array.Array;
 import com.ss.rlib.common.util.array.ConcurrentArray;
@@ -63,7 +65,7 @@ private static void removeExpiredPackets(@NotNull Array<PendingPublish> publishe
     private final @NotNull String clientId;
     private final @NotNull ConcurrentArray<PendingPublish> pendingPublishes;
     private final @NotNull AtomicInteger packetIdGenerator;
-    private final @Getter @NotNull ConcurrentArray<SubscribeTopicFilter> topicFilters;
+    private final @NotNull ConcurrentArray<SubscribeTopicFilter> topicFilters;
 
     private volatile @Getter @Setter long expirationTime = -1;
 
@@ -165,6 +167,29 @@ public void updatePendingPacket(
         }
     }
 
+    @Override
+    public <F, S> void forEachTopicFilter(
+        @NotNull F first,
+        @NotNull S second,
+        @NotNull NotNullTripleConsumer<F, S, SubscribeTopicFilter> consumer
+    ) {
+        topicFilters.forEachInReadLock(first, second, consumer);
+    }
+
+    @Override
+    public void addSubscriber(@NotNull SubscribeTopicFilter subscribe) {
+        topicFilters.runInWriteLock(subscribe, Collection::add);
+    }
+
+    @Override
+    public void removeSubscriber(@NotNull TopicFilter topicFilter) {
+        topicFilters.removeIfConvertedInWriteLock(
+            topicFilter,
+            SubscribeTopicFilter::getTopicFilter,
+            Object::equals
+        );
+    }
+
     @Override
     public void clear() {
         pendingPublishes.runInWriteLock(Collection::clear);
diff --git a/src/main/java/com/ss/mqtt/broker/model/topic/AbstractTopic.java b/src/main/java/com/ss/mqtt/broker/model/topic/AbstractTopic.java
@@ -32,7 +32,6 @@ static void checkTopic(@NotNull String topic) {
     }
 
     AbstractTopic(@NotNull String topicName) {
-
         length = topicName.length();
         segments = topicName.split(DELIMITER);
         rawTopic = topicName;
diff --git a/src/main/java/com/ss/mqtt/broker/model/topic/TopicFilter.java b/src/main/java/com/ss/mqtt/broker/model/topic/TopicFilter.java
@@ -4,7 +4,7 @@
 
 public class TopicFilter extends AbstractTopic {
 
-    public static TopicFilter from(@NotNull String topicFilter) {
+    public static @NotNull TopicFilter from(@NotNull String topicFilter) {
         checkTopic(topicFilter);
         int multiPos = topicFilter.indexOf(MULTI_LEVEL_WILDCARD);
         if (multiPos != -1 && multiPos != topicFilter.length() - 1) {
diff --git a/src/main/java/com/ss/mqtt/broker/model/topic/TopicSubscribers.java b/src/main/java/com/ss/mqtt/broker/model/topic/TopicSubscribers.java
@@ -10,6 +10,7 @@
 import com.ss.rlib.common.util.array.ConcurrentArray;
 import com.ss.rlib.common.util.dictionary.ConcurrentObjectDictionary;
 import com.ss.rlib.common.util.dictionary.ObjectDictionary;
+import lombok.Getter;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -60,19 +61,8 @@ private static boolean removeDuplicateWithLowerQoS(
         return topicSubscribers;
     }
 
-    private @Nullable ConcurrentObjectDictionary<String, TopicSubscribers> topicSubscribers;
-    private @Nullable ConcurrentArray<Subscriber> subscribers;
-
-    public void restoreSubscribers(
-        @NotNull MqttClient mqttClient,
-        @NotNull ConcurrentArray<SubscribeTopicFilter> topicFilters
-    ) {
-        topicFilters.forEachInReadLock(
-            this,
-            mqttClient,
-            TopicSubscribers::addSubscriber
-        );
-    }
+    private volatile @Getter @Nullable ConcurrentObjectDictionary<String, TopicSubscribers> topicSubscribers;
+    private volatile @Getter @Nullable ConcurrentArray<Subscriber> subscribers;
 
     public void addSubscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter subscribe) {
         addSubscriber(0, subscribe.getTopicFilter(), new Subscriber(mqttClient, subscribe));
@@ -92,16 +82,8 @@ private void addSubscriber(int level, @NotNull TopicFilter topicFilter, @NotNull
         }
     }
 
-    public void cleanSubscribers(
-        @NotNull MqttClient mqttClient,
-        @NotNull ConcurrentArray<SubscribeTopicFilter> topicFilters
-    ) {
-        topicFilters.forEachConvertedInReadLock(
-            this,
-            mqttClient,
-            SubscribeTopicFilter::getTopicFilter,
-            TopicSubscribers::removeSubscriber
-        );
+    public boolean removeSubscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter subscribe) {
+        return removeSubscriber(mqttClient, subscribe.getTopicFilter());
     }
 
     public boolean removeSubscriber(@NotNull MqttClient mqttClient, @NotNull TopicFilter topicFilter) {
@@ -192,12 +174,4 @@ private void processSegment(
         }
         return subscribers;
     }
-
-    private synchronized @Nullable ConcurrentObjectDictionary<String, TopicSubscribers> getTopicSubscribers() {
-        return topicSubscribers;
-    }
-
-    private synchronized @Nullable ConcurrentArray<Subscriber> getSubscribers() {
-        return subscribers;
-    }
 }
diff --git a/src/main/java/com/ss/mqtt/broker/service/SubscriptionService.java b/src/main/java/com/ss/mqtt/broker/service/SubscriptionService.java
@@ -52,4 +52,8 @@ public interface SubscriptionService {
         @NotNull MqttClient mqttClient,
         @NotNull Array<TopicFilter> topicFilters
     );
+
+    void cleanSubscriptions(@NotNull MqttClient mqttClient, @NotNull MqttSession mqttSession);
+
+    void restoreSubscriptions(@NotNull MqttClient mqttClient, @NotNull MqttSession mqttSession);
 }
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptionService.java b/src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptionService.java
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectSubscribePublishTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectSubscribePublishTest.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/service/SubscribtionServiceTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/service/SubscribtionServiceTest.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/NetworkUnitSpecification.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/NetworkUnitSpecification.groovy

Original file line number	Diff line number	Diff line change
`@@ -94,11 +94,7 @@ allprojects {`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`tasks.withType(GroovyCompile) {`
`97`		`- options.forkOptions.jvmArgs << "--enable-preview"`
`98`		`- }`
`99`		`-`
`100`		`- tasks.withType(JavaCompile) {`
`101`		`- options.compilerArgs.add("--enable-preview")`
	`97`	`+ options.forkOptions.jvmArgs += "--enable-preview"`
`102`	`98`	`}`
`103`	`99`
`104`	`100`	`processResources {`
Original file line number	Diff line number	Diff line change
`@@ -25,11 +25,11 @@ public Subscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter`
`25`	`25`	`this.subscribeTopicFilter = topicFilter;`
`26`	`26`	`}`
`27`	`27`
`28`		`- public QoS getQos() {`
	`28`	`+ public @NotNull QoS getQos() {`
`29`	`29`	`return subscribeTopicFilter.getQos();`
`30`	`30`	`}`
`31`	`31`
`32`		`- public TopicFilter getTopicFilter() {`
	`32`	`+ public @NotNull TopicFilter getTopicFilter() {`
`33`	`33`	`return subscribeTopicFilter.getTopicFilter();`
`34`	`34`	`}`
`35`	`35`