[broker-12] add concurrent access support to subscription and authentication services

crazyrokr · crazyrokr · commit d0af0b31f375 · 2019-11-08T17:52:22.000+03:00
diff --git a/src/main/java/com/ss/mqtt/broker/model/SubscribeTopicFilter.java b/src/main/java/com/ss/mqtt/broker/model/SubscribeTopicFilter.java
@@ -8,9 +8,9 @@
 public class SubscribeTopicFilter {
 
     /**
-     * The subscriber's topic filter.
+     * The subscriber's topic name.
      */
-    private final String topicFilter;
+    private final String topicName;
 
     /**
      * Maximum QoS field. This gives the maximum QoS level at which the Server
diff --git a/src/main/java/com/ss/mqtt/broker/service/SubscriptionService.java b/src/main/java/com/ss/mqtt/broker/service/SubscriptionService.java
@@ -2,6 +2,7 @@
 
 import com.ss.mqtt.broker.model.SubscribeAckReasonCode;
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
+import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.model.UnsubscribeAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.rlib.common.util.array.Array;
@@ -42,5 +43,5 @@ public interface SubscriptionService {
      * @param topicName topic name
      * @return array of topic subscribers
      */
-    @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName);
+    @NotNull Array<Subscriber> getSubscribers(@NotNull String topicName);
 }
diff --git a/src/main/java/com/ss/mqtt/broker/service/Subscriptions.java b/src/main/java/com/ss/mqtt/broker/service/Subscriptions.java
@@ -2,6 +2,7 @@
 
 import com.ss.mqtt.broker.model.SubscribeAckReasonCode;
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
+import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.model.UnsubscribeAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.rlib.common.util.array.Array;
@@ -18,7 +19,7 @@ public interface Subscriptions {
      * @param topicName topic name on which subscribers should be returned
      * @return array of MQTT clients
      */
-    @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName);
+    @NotNull Array<Subscriber> getSubscribers(@NotNull String topicName);
 
     /**
      * Returns result of subscription adding
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/AbstractCredentialSource.java b/src/main/java/com/ss/mqtt/broker/service/impl/AbstractCredentialSource.java
@@ -1,22 +1,28 @@
 package com.ss.mqtt.broker.service.impl;
 
 import com.ss.mqtt.broker.service.CredentialSource;
+import com.ss.rlib.common.util.dictionary.ConcurrentObjectDictionary;
+import com.ss.rlib.common.util.dictionary.Dictionary;
+import com.ss.rlib.common.util.dictionary.DictionaryFactory;
+import com.ss.rlib.common.util.dictionary.ObjectDictionary;
 import org.jetbrains.annotations.NotNull;
 import reactor.core.publisher.Mono;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
 
 public abstract class AbstractCredentialSource implements CredentialSource {
 
-    private final Map<String, byte[]> credentials = new HashMap<>();
+    private final ConcurrentObjectDictionary<String, byte[]> credentials =
+        DictionaryFactory.newConcurrentStampedLockObjectDictionary();
 
     abstract void init();
 
-    void putCredentials(@NotNull Object user, @NotNull Object pass) {
-        credentials.put(user.toString(), pass.toString().getBytes(StandardCharsets.UTF_8));
+    void putAll(@NotNull Dictionary<String, byte[]> creds) {
+        credentials.runInWriteLock(creds, Dictionary::put);
+    }
+
+    void put(@NotNull String user, @NotNull byte[] pass) {
+        credentials.runInWriteLock(user, pass, ObjectDictionary::put);
     }
 
     @Override
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/FileCredentialsSource.java b/src/main/java/com/ss/mqtt/broker/service/impl/FileCredentialsSource.java
@@ -1,10 +1,13 @@
 package com.ss.mqtt.broker.service.impl;
 
 import com.ss.mqtt.broker.exception.CredentialsSourceException;
+import com.ss.rlib.common.util.dictionary.Dictionary;
+import com.ss.rlib.common.util.dictionary.DictionaryCollectors;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Properties;
 
 public class FileCredentialsSource extends AbstractCredentialSource {
@@ -25,7 +28,16 @@ void init() {
         try {
             var credentialsProperties = new Properties();
             credentialsProperties.load(new FileInputStream(credentialUrl.getPath()));
-            credentialsProperties.forEach(this::putCredentials);
+
+            Dictionary<String, byte[]> creds = credentialsProperties.entrySet()
+                .stream()
+                .collect(DictionaryCollectors.toObjectDictionary(
+                    entry -> entry.getKey().toString(),
+                    entry -> entry.getValue().toString().getBytes(StandardCharsets.UTF_8)
+                ));
+
+            putAll(creds);
+
         } catch (IOException e) {
             throw new CredentialsSourceException(e);
         }
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/SimplePublishingService.java b/src/main/java/com/ss/mqtt/broker/service/impl/SimplePublishingService.java
@@ -1,7 +1,7 @@
 package com.ss.mqtt.broker.service.impl;
 
 import com.ss.mqtt.broker.model.PublishAckReasonCode;
-import com.ss.mqtt.broker.network.client.MqttClient;
+import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import com.ss.mqtt.broker.service.PublishingService;
 import com.ss.mqtt.broker.service.SubscriptionService;
@@ -17,10 +17,11 @@ public class SimplePublishingService implements PublishingService {
     private final @NotNull SubscriptionService subscriptionService;
 
     private static @NotNull PublishAckReasonCode send(
-        @NotNull MqttClient mqttClient,
+        @NotNull Subscriber subscriber,
         @NotNull PublishInPacket publish
     ) {
 
+        var mqttClient = subscriber.getMqttClient();
         mqttClient.send(mqttClient.getPacketOutFactory().newPublish(
             mqttClient,
             publish.getPacketId(),
@@ -50,7 +51,7 @@ public class SimplePublishingService implements PublishingService {
         }
 
         var success = subscribers.stream()
-            .map(targetMqttClient -> send(targetMqttClient, publish))
+            .map(subscriber -> send(subscriber, publish))
             .allMatch(ackReasonCode -> ackReasonCode.equals(PublishAckReasonCode.SUCCESS));
 
         return success ? PublishAckReasonCode.SUCCESS : PublishAckReasonCode.UNSPECIFIED_ERROR;
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
@@ -2,6 +2,7 @@
 
 import com.ss.mqtt.broker.model.SubscribeAckReasonCode;
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
+import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.model.UnsubscribeAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.service.SubscriptionService;
@@ -40,7 +41,7 @@ public class SimpleSubscriptionService implements SubscriptionService {
     }
 
     @Override
-    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName) {
+    public @NotNull Array<Subscriber> getSubscribers(@NotNull String topicName) {
         return subscriptions.getSubscribers(topicName);
     }
 }
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptions.java b/src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptions.java
@@ -6,50 +6,73 @@
 import com.ss.mqtt.broker.model.UnsubscribeAckReasonCode;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.service.Subscriptions;
+import com.ss.rlib.common.function.NotNullSupplier;
 import com.ss.rlib.common.util.array.Array;
-import com.ss.rlib.common.util.array.ArrayCollectors;
-import com.ss.rlib.common.util.array.impl.FastArraySet;
+import com.ss.rlib.common.util.array.ConcurrentArray;
+import com.ss.rlib.common.util.dictionary.ConcurrentObjectDictionary;
+import com.ss.rlib.common.util.dictionary.DictionaryFactory;
+import com.ss.rlib.common.util.dictionary.ObjectDictionary;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.HashMap;
-import java.util.Map;
-
 /**
  * Simple container of subscriptions
  */
 public class SimpleSubscriptions implements Subscriptions {
 
-    private final @NotNull Map<String, Array<Subscriber>> subscriptions = new HashMap<>();
+    private final static @NotNull NotNullSupplier<ConcurrentArray<Subscriber>> SUBSCRIBER_ARRAY_SUPPLIER =
+        ConcurrentArray.supplier(Subscriber.class);
+
+    private final @NotNull ConcurrentObjectDictionary<String, ConcurrentArray<Subscriber>> subscriptions =
+        DictionaryFactory.newConcurrentStampedLockObjectDictionary();
 
-    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName) {
-        return subscriptions.get(topicName)
-            .stream()
-            .map(Subscriber::getMqttClient)
-            .collect(ArrayCollectors.toArray(MqttClient.class));
+    public @NotNull Array<Subscriber> getSubscribers(@NotNull String topicName) {
+
+        var subscribers = subscriptions.getInReadLock(topicName, ObjectDictionary::get);
+        if (subscribers == null) {
+            return Array.empty();
+        }
+
+        //noinspection ConstantConditions
+        return subscribers.getInReadLock(Array::of);
     }
 
     public @NotNull SubscribeAckReasonCode addSubscription(
         @NotNull SubscribeTopicFilter topicFilter,
         @NotNull MqttClient mqttClient
     ) {
         var subscriber = new Subscriber(mqttClient, topicFilter);
-        var subscribers = subscriptions.computeIfAbsent(
-            topicFilter.getTopicFilter(),
-            key -> new FastArraySet<>(Subscriber.class)
-        );
-        subscribers.add(subscriber);
+        var subscribers = subscriptions.getInReadLock(topicFilter.getTopicName(), ObjectDictionary::get);
+
+        if (subscribers == null) {
+            subscribers = subscriptions.getInWriteLock(
+                topicFilter.getTopicName(),
+                SUBSCRIBER_ARRAY_SUPPLIER,
+                ObjectDictionary::getOrCompute
+            );
+        }
+
+        //noinspection ConstantConditions
+        subscribers.runInWriteLock(subscriber, Array::add);
+
         return topicFilter.getQos().getSubscribeAckReasonCode();
     }
 
     public @NotNull UnsubscribeAckReasonCode removeSubscription(
         @NotNull String topicName,
         @NotNull MqttClient mqttClient
     ) {
-        var subscribers = subscriptions.getOrDefault(topicName, Array.empty());
-        if (subscribers.removeIf(subscriber -> mqttClient.equals(subscriber.getMqttClient()))) {
-            return UnsubscribeAckReasonCode.SUCCESS;
-        } else {
+        var subscribers = subscriptions.getInReadLock(topicName, ObjectDictionary::get);
+
+        if (subscribers == null) {
             return UnsubscribeAckReasonCode.NO_SUBSCRIPTION_EXISTED;
+        } else {
+            //noinspection ConstantConditions
+            boolean removed = subscribers.getInWriteLock(
+                mqttClient,
+                (subs, client) -> subs.removeIf(subscriber -> client.equals(subscriber.getMqttClient()))
+            );
+
+            return removed ? UnsubscribeAckReasonCode.SUCCESS : UnsubscribeAckReasonCode.NO_SUBSCRIPTION_EXISTED;
         }
     }
 }
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/in/SubscribeInPacketTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/in/SubscribeInPacketTest.groovy
@@ -29,12 +29,12 @@ class SubscribeInPacketTest extends BaseInPacketTest {
             result
             packet.topicFilters.size() == 2
             packet.topicFilters.get(0).getQos() == QoS.AT_LEAST_ONCE_DELIVERY
-            packet.topicFilters.get(0).getTopicFilter() == topicFilter
+            packet.topicFilters.get(0).getTopicName() == topicFilter
             packet.topicFilters.get(0).isNoLocal()
             !packet.topicFilters.get(0).isRetainAsPublished()
             packet.topicFilters.get(0).getRetainHandling() == SubscribeRetainHandling.SEND_AT_THE_TIME_OF_SUBSCRIBE
             packet.topicFilters.get(1).getQos() == QoS.EXACTLY_ONCE_DELIVERY
-            packet.topicFilters.get(1).getTopicFilter() == topicFilter2
+            packet.topicFilters.get(1).getTopicName() == topicFilter2
             packet.topicFilters.get(1).isNoLocal()
             !packet.topicFilters.get(1).isRetainAsPublished()
             packet.topicFilters.get(1).getRetainHandling() == SubscribeRetainHandling.SEND_AT_THE_TIME_OF_SUBSCRIBE
@@ -69,12 +69,12 @@ class SubscribeInPacketTest extends BaseInPacketTest {
             result
             packet.topicFilters.size() == 2
             packet.topicFilters.get(0).getQos() == QoS.AT_LEAST_ONCE_DELIVERY
-            packet.topicFilters.get(0).getTopicFilter() == topicFilter
+            packet.topicFilters.get(0).getTopicName() == topicFilter
             !packet.topicFilters.get(0).isNoLocal()
             packet.topicFilters.get(0).isRetainAsPublished()
             packet.topicFilters.get(0).getRetainHandling() == SubscribeRetainHandling.SEND_AT_THE_TIME_OF_SUBSCRIBE
             packet.topicFilters.get(1).getQos() == QoS.EXACTLY_ONCE_DELIVERY
-            packet.topicFilters.get(1).getTopicFilter() == topicFilter2
+            packet.topicFilters.get(1).getTopicName() == topicFilter2
             packet.topicFilters.get(1).isNoLocal()
             !packet.topicFilters.get(1).isRetainAsPublished()
             packet.topicFilters.get(1).getRetainHandling() == SubscribeRetainHandling.SEND_AT_SUBSCRIBE_ONLY_IF_THE_SUBSCRIPTION_DOES_NOT_CURRENTLY_EXIST
@@ -98,12 +98,12 @@ class SubscribeInPacketTest extends BaseInPacketTest {
             result
             packet.topicFilters.size() == 2
             packet.topicFilters.get(0).getQos() == QoS.AT_LEAST_ONCE_DELIVERY
-            packet.topicFilters.get(0).getTopicFilter() == topicFilter
+            packet.topicFilters.get(0).getTopicName() == topicFilter
             !packet.topicFilters.get(0).isNoLocal()
             !packet.topicFilters.get(0).isRetainAsPublished()
             packet.topicFilters.get(0).getRetainHandling() == SubscribeRetainHandling.SEND_AT_THE_TIME_OF_SUBSCRIBE
             packet.topicFilters.get(1).getQos() == QoS.EXACTLY_ONCE_DELIVERY
-            packet.topicFilters.get(1).getTopicFilter() == topicFilter2
+            packet.topicFilters.get(1).getTopicName() == topicFilter2
             !packet.topicFilters.get(1).isNoLocal()
             !packet.topicFilters.get(1).isRetainAsPublished()
             packet.topicFilters.get(1).getRetainHandling() == SubscribeRetainHandling.SEND_AT_THE_TIME_OF_SUBSCRIBE

Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`
`3`	`3`	`import com.ss.mqtt.broker.model.SubscribeAckReasonCode;`
`4`	`4`	`import com.ss.mqtt.broker.model.SubscribeTopicFilter;`
	`5`	`+import com.ss.mqtt.broker.model.Subscriber;`
`5`	`6`	`import com.ss.mqtt.broker.model.UnsubscribeAckReasonCode;`
`6`	`7`	`import com.ss.mqtt.broker.network.client.MqttClient;`
`7`	`8`	`import com.ss.mqtt.broker.service.SubscriptionService;`
`@@ -40,7 +41,7 @@ public class SimpleSubscriptionService implements SubscriptionService {`
`40`	`41`	`}`
`41`	`42`
`42`	`43`	`@Override`
`43`		`- public @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName) {`
	`44`	`+ public @NotNull Array<Subscriber> getSubscribers(@NotNull String topicName) {`
`44`	`45`	`return subscriptions.getSubscribers(topicName);`
`45`	`46`	`}`
`46`	`47`	`}`