Merge pull request #38 from JavaSaBr/feature-broker-23

Feature broker 23: Shared subscriptions

build.gradle

@@ -20,8 +20,8 @@ allprojects {
     apply plugin: "groovy"
     apply plugin: "org.springframework.boot"
 
-    sourceCompatibility = JavaVersion.VERSION_13
-    targetCompatibility = JavaVersion.VERSION_13
+    sourceCompatibility = JavaVersion.VERSION_14
+    targetCompatibility = JavaVersion.VERSION_14
 
     repositories {
         jcenter()
@@ -38,8 +38,8 @@ allprojects {
         springVersion = '5.1.6.RELEASE'
         junitJupiterVersion = "5.5.2"
         testcontainersVersion = "1.12.1"
-        groovyVersion = "2.5.8"
-        spockVersion = "1.2-groovy-2.5"
+        groovyVersion = "3.0.4"
+        spockVersion = "2.0-M3-groovy-3.0"
         projectReactorVersion = "3.3.0.RELEASE"
         byteBuddyVersion = "1.10.2"
         objenesisVersion = "3.1"
@@ -118,7 +118,7 @@ task buildSingleArtifactWithoutTests(type: GradleBuild) {
 }
 
 wrapper {
-    gradleVersion = '6.0'
+    gradleVersion = '6.6-milestone-1'
     distributionType = Wrapper.DistributionType.ALL
 }
 

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.0-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.6-milestone-1-all.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradlew

@@ -82,6 +82,7 @@ esac
 
 CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
 
+
 # Determine the Java command to use to start the JVM.
 if [ -n "$JAVA_HOME" ] ; then
     if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
@@ -129,6 +130,7 @@ fi
 if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+
     JAVACMD=`cygpath --unix "$JAVACMD"`
 
     # We build the pattern for arguments to be converted via cygpath

gradlew.bat

@@ -29,6 +29,9 @@ if "%DIRNAME%" == "" set DIRNAME=.
 set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
 @rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
 set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
 
@@ -81,6 +84,7 @@ set CMD_LINE_ARGS=%*
 
 set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
 
+
 @rem Execute Gradle
 "%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
 

src/main/java/com/ss/mqtt/broker/config/MqttNetworkConfig.java

@@ -1,13 +1,12 @@
 package com.ss.mqtt.broker.config;
 
 import com.ss.mqtt.broker.handler.client.MqttClientReleaseHandler;
-import com.ss.mqtt.broker.handler.packet.in.*;
+import com.ss.mqtt.broker.handler.packet.in.PacketInHandler;
 import com.ss.mqtt.broker.model.MqttPropertyConstants;
 import com.ss.mqtt.broker.model.QoS;
 import com.ss.mqtt.broker.network.MqttConnection;
 import com.ss.mqtt.broker.network.client.ExternalMqttClient;
 import com.ss.mqtt.broker.network.client.InternalMqttClient;
-import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.network.client.MqttClient.UnsafeMqttClient;
 import com.ss.rlib.network.BufferAllocator;
 import com.ss.rlib.network.Network;

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

@@ -9,8 +9,10 @@ public abstract class AbstractPacketHandler<C extends UnsafeMqttClient, R extend
 
     @Override
     public void handle(@NotNull UnsafeMqttClient client, @NotNull MqttReadablePacket packet) {
+        //noinspection unchecked
         handleImpl((C) client, (R) packet);
     }
 
     protected abstract void handleImpl(@NotNull C client, @NotNull R packet);
 }
+

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

@@ -20,6 +20,6 @@ protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull DisconnectI
             log.error("Disconnect client {} by error reason {}", client, reasonCode);
         }
 
-        client.release().subscribe();
+        client.getConnection().close();
     }
 }

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

@@ -1,19 +1,37 @@
 package com.ss.mqtt.broker.handler.packet.in;
 
+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;
 import com.ss.mqtt.broker.network.packet.in.SubscribeInPacket;
 import com.ss.mqtt.broker.service.SubscriptionService;
 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));
+        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());
+        }
     }
 }

src/main/java/com/ss/mqtt/broker/handler/publish/in/AbstractPublishInHandler.java

@@ -3,7 +3,7 @@
 import com.ss.mqtt.broker.handler.publish.out.PublishOutHandler;
 import com.ss.mqtt.broker.model.ActionResult;
 import com.ss.mqtt.broker.model.QoS;
-import com.ss.mqtt.broker.model.Subscriber;
+import com.ss.mqtt.broker.model.SingleSubscriber;
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import com.ss.mqtt.broker.service.SubscriptionService;
@@ -25,7 +25,7 @@ public void handle(@NotNull MqttClient client, @NotNull PublishInPacket packet)
     }
 
     private @NotNull ActionResult sendToSubscriber(
-        @NotNull Subscriber subscriber,
+        @NotNull SingleSubscriber subscriber,
         @NotNull PublishInPacket packet
     ) {
         return publishOutHandler(subscriber.getQos()).handle(packet, subscriber);

src/main/java/com/ss/mqtt/broker/handler/publish/out/AbstractPublishOutHandler.java

@@ -8,7 +8,7 @@
 abstract class AbstractPublishOutHandler implements PublishOutHandler {
 
     @Override
-    public @NotNull ActionResult handle(@NotNull PublishInPacket packet, @NotNull Subscriber subscriber) {
+    public @NotNull ActionResult handle(@NotNull PublishInPacket packet, @NotNull SingleSubscriber subscriber) {
 
         var client = subscriber.getMqttClient();
         var session = client.getSession();
@@ -30,7 +30,7 @@ abstract class AbstractPublishOutHandler implements PublishOutHandler {
 
     protected abstract @NotNull QoS getQoS();
 
-    protected void sendPublish(
+    void sendPublish(
         @NotNull MqttClient client,
         @NotNull PublishInPacket packet,
         int packetId,

src/main/java/com/ss/mqtt/broker/handler/publish/out/PublishOutHandler.java

@@ -1,6 +1,7 @@
 package com.ss.mqtt.broker.handler.publish.out;
 
 import com.ss.mqtt.broker.model.ActionResult;
+import com.ss.mqtt.broker.model.SingleSubscriber;
 import com.ss.mqtt.broker.model.Subscriber;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import org.jetbrains.annotations.NotNull;
@@ -10,5 +11,5 @@
  */
 public interface PublishOutHandler {
 
-    @NotNull ActionResult handle(@NotNull PublishInPacket packet, @NotNull Subscriber subscriber);
+    @NotNull ActionResult handle(@NotNull PublishInPacket packet, @NotNull SingleSubscriber subscriber);
 }

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

@@ -62,10 +62,10 @@ public enum MqttVersion {
         return availableVersions[level];
     }
 
-    private final String name;
-    private final byte[] nameInBytes;
+    private final @NotNull MqttPacketOutFactory packetOutFactory;
+    private final byte @NotNull [] nameInBytes;
+    private final @NotNull String name;
     private final byte version;
-    private final MqttPacketOutFactory packetOutFactory;
 
     MqttVersion(@NotNull String name, int version, @NotNull MqttPacketOutFactory packetOutFactory) {
         this.name = name;

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

@@ -64,8 +64,8 @@ public enum PacketProperty {
     }
 
     private final @Getter byte id;
-    private final @Getter PacketDataType dataType;
-    private final Object defaultValue;
+    private final @Getter @NotNull PacketDataType dataType;
+    private final @Nullable Object defaultValue;
 
     PacketProperty(int id, @NotNull PacketDataType dataType) {
         this(id, dataType, null);

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

@@ -0,0 +1,50 @@
+package com.ss.mqtt.broker.model;
+
+import com.ss.mqtt.broker.model.topic.SharedTopicFilter;
+import com.ss.mqtt.broker.network.client.MqttClient;
+import com.ss.rlib.common.util.array.Array;
+import com.ss.rlib.common.util.array.ConcurrentArray;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.Collection;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SharedSubscriber implements Subscriber {
+
+    private static @NotNull SingleSubscriber next(@NotNull Array<SingleSubscriber> subscribers, int current) {
+        return subscribers.get(current % subscribers.size());
+    }
+
+    private final @NotNull SharedTopicFilter topicFilter;
+    private final @NotNull ConcurrentArray<SingleSubscriber> subscribers;
+    private final @NotNull AtomicInteger current;
+
+    public SharedSubscriber(@NotNull SubscribeTopicFilter topic) {
+        subscribers = ConcurrentArray.ofType(Subscriber.class);
+        current = new AtomicInteger(0);
+        topicFilter = (SharedTopicFilter) topic.getTopicFilter();
+    }
+
+    public @NotNull SingleSubscriber getSubscriber() {
+        //noinspection ConstantConditions
+        return subscribers.getInReadLock(current.getAndIncrement(), SharedSubscriber::next);
+    }
+
+    public void addSubscriber(@NotNull SingleSubscriber client) {
+        subscribers.runInWriteLock(client, Array::add);
+    }
+
+    public boolean removeSubscriber(@NotNull MqttClient client) {
+        return subscribers.removeIfConvertedInWriteLock(client, SingleSubscriber::getMqttClient, Objects::equals);
+    }
+
+    public int size() {
+        //noinspection ConstantConditions
+        return subscribers.getInReadLock(Collection::size);
+    }
+
+    public @NotNull String getGroup() {
+        return topicFilter.getGroup();
+    }
+}

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

@@ -0,0 +1,21 @@
+package com.ss.mqtt.broker.model;
+
+import com.ss.mqtt.broker.network.client.MqttClient;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.jetbrains.annotations.NotNull;
+
+@ToString
+@EqualsAndHashCode(of = "mqttClient")
+@RequiredArgsConstructor
+public class SingleSubscriber implements Subscriber {
+
+    private final @Getter @NotNull MqttClient mqttClient;
+    private final @NotNull SubscribeTopicFilter subscribe;
+
+    public @NotNull QoS getQos() {
+        return subscribe.getQos();
+    }
+}

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

@@ -1,7 +1,8 @@
 package com.ss.mqtt.broker.model;
 
-import lombok.EqualsAndHashCode;
+import static com.ss.mqtt.broker.util.TopicUtils.buildTopicFilter;
 import com.ss.mqtt.broker.model.topic.TopicFilter;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
@@ -43,7 +44,7 @@ public class SubscribeTopicFilter {
     private final boolean retainAsPublished;
 
     public SubscribeTopicFilter(@NotNull String topicFilter, @NotNull QoS qos) {
-        this(TopicFilter.from(topicFilter), qos, SubscribeRetainHandling.SEND, true, true);
+        this(buildTopicFilter(topicFilter), qos, SubscribeRetainHandling.SEND, true, true);
     }
 
     public SubscribeTopicFilter(@NotNull TopicFilter topicFilter, @NotNull QoS qos) {

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

@@ -1,36 +1,3 @@
 package com.ss.mqtt.broker.model;
 
-import com.ss.mqtt.broker.model.topic.TopicFilter;
-import com.ss.mqtt.broker.network.client.MqttClient;
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.ToString;
-import org.jetbrains.annotations.NotNull;
-
-@ToString
-@EqualsAndHashCode(of = "mqttClient")
-public class Subscriber {
-
-    private final @Getter @NotNull MqttClient mqttClient;
-    private final @NotNull SubscribeTopicFilter subscribeTopicFilter;
-
-    /**
-     * Creates subscriber
-     *
-     * @param mqttClient  MQTT client which will become a subscriber
-     * @param topicFilter topic filter that MQTT client subscribes to
-     */
-    public Subscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter topicFilter) {
-        this.mqttClient = mqttClient;
-        this.subscribeTopicFilter = topicFilter;
-    }
-
-    public @NotNull QoS getQos() {
-        return subscribeTopicFilter.getQos();
-    }
-
-    public @NotNull TopicFilter getTopicFilter() {
-        return subscribeTopicFilter.getTopicFilter();
-    }
-
-}
+public interface Subscriber {}

src/main/java/com/ss/mqtt/broker/model/data/type/StringPair.java

@@ -4,13 +4,14 @@
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
+import org.jetbrains.annotations.NotNull;
 
 @Getter
 @ToString
 @EqualsAndHashCode
 @RequiredArgsConstructor
 public class StringPair {
 
-    private final String name;
-    private final String value;
+    private final @NotNull String name;
+    private final @NotNull String value;
 }