Merge pull request #16 from JavaSaBr/feature-broker-11

Implement Mqtt Session Service

Author: crazyrokr

README.md

@@ -3,15 +3,15 @@
 JMQTT-Broker is an open source (GPL v3) Java based MQTT Broker with network implementation based on Java.NIO.2
 
 ## Dependencies for building
-### java 11+
+### java 13+
 ##### Ubuntu:
 ```bash
 sudo wget -qO - https://adoptopenjdk.jfrog.io/adoptopenjdk/api/gpg/key/public | sudo apt-key add -
 sudo add-apt-repository --yes https://adoptopenjdk.jfrog.io/adoptopenjdk/deb/
-sudo apt-get install adoptopenjdk-11-hotspot
+sudo apt-get install adoptopenjdk-13-hotspot
 ```
-##### Windows: [Windows x64 installer](https://github.com/AdoptOpenJDK/openjdk11-binaries/releases/download/jdk-11.0.4%2B11/OpenJDK11U-jdk_x64_windows_hotspot_11.0.4_11.msi)
-##### MacOS: [MacOS x64 installer](https://github.com/AdoptOpenJDK/openjdk11-binaries/releases/download/jdk-11.0.4%2B11.4/OpenJDK11U-jdk_x64_mac_hotspot_11.0.4_11.pkg)
+##### Windows: [Windows x64 installer](https://github.com/AdoptOpenJDK/openjdk13-binaries/releases/download/jdk-13.0.1%2B9/OpenJDK13U-jdk_x64_windows_hotspot_13.0.1_9.msi)
+##### MacOS: [MacOS x64 installer](https://github.com/AdoptOpenJDK/openjdk13-binaries/releases/download/jdk-13.0.1%2B9/OpenJDK13U-jdk_x64_mac_hotspot_13.0.1_9.pkg)
 ### Docker
 ##### Ubuntu: [installation guide](https://docs.docker.com/install/linux/docker-ce/ubuntu)
 ##### Windows: [installation guide](https://docs.docker.com/docker-for-windows/install)

build.gradle

@@ -20,8 +20,8 @@ allprojects {
     apply plugin: "groovy"
     apply plugin: "org.springframework.boot"
 
-    sourceCompatibility = JavaVersion.VERSION_11
-    targetCompatibility = JavaVersion.VERSION_11
+    sourceCompatibility = JavaVersion.VERSION_13
+    targetCompatibility = JavaVersion.VERSION_13
 
     repositories {
         jcenter()
@@ -107,7 +107,7 @@ task buildSingleArtifactWithoutTests(type: GradleBuild) {
 }
 
 wrapper {
-    gradleVersion = '5.6.1'
+    gradleVersion = '6.0'
     distributionType = Wrapper.DistributionType.ALL
 }
 

gradle/wrapper/gradle-wrapper.jar

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

gradle/wrapper/gradle-wrapper.properties

@@ -7,7 +7,7 @@
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
-#      http://www.apache.org/licenses/LICENSE-2.0
+#      https://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
@@ -125,8 +125,8 @@ if $darwin; then
     GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
 fi
 
-# For Cygwin, switch paths to Windows format before running java
-if $cygwin ; then
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
     JAVACMD=`cygpath --unix "$JAVACMD"`
@@ -154,19 +154,19 @@ if $cygwin ; then
         else
             eval `echo args$i`="\"$arg\""
         fi
-        i=$((i+1))
+        i=`expr $i + 1`
     done
     case $i in
-        (0) set -- ;;
-        (1) set -- "$args0" ;;
-        (2) set -- "$args0" "$args1" ;;
-        (3) set -- "$args0" "$args1" "$args2" ;;
-        (4) set -- "$args0" "$args1" "$args2" "$args3" ;;
-        (5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
-        (6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
-        (7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
-        (8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
-        (9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
+        0) set -- ;;
+        1) set -- "$args0" ;;
+        2) set -- "$args0" "$args1" ;;
+        3) set -- "$args0" "$args1" "$args2" ;;
+        4) set -- "$args0" "$args1" "$args2" "$args3" ;;
+        5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
+        6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
+        7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
+        8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
+        9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
     esac
 fi
 
@@ -175,14 +175,9 @@ save () {
     for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
     echo " "
 }
-APP_ARGS=$(save "$@")
+APP_ARGS=`save "$@"`
 
 # Collect all arguments for the java command, following the shell quoting and substitution rules
 eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS"
 
-# by default we should be in the correct project dir, but when run from Finder on Mac, the cwd is wrong
-if [ "$(uname)" = "Darwin" ] && [ "$HOME" = "$PWD" ]; then
-  cd "$(dirname "$0")"
-fi
-
 exec "$JAVACMD" "$@"

gradlew

@@ -5,7 +5,7 @@
 @rem you may not use this file except in compliance with the License.
 @rem You may obtain a copy of the License at
 @rem
-@rem      http://www.apache.org/licenses/LICENSE-2.0
+@rem      https://www.apache.org/licenses/LICENSE-2.0
 @rem
 @rem Unless required by applicable law or agreed to in writing, software
 @rem distributed under the License is distributed on an "AS IS" BASIS,

gradlew.bat

@@ -3,8 +3,10 @@
 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.MqttClientReleaseHandler;
 import com.ss.mqtt.broker.network.client.UnsafeMqttClient;
 import com.ss.mqtt.broker.network.client.impl.DeviceMqttClient;
+import com.ss.mqtt.broker.network.client.impl.DeviceMqttClientReleaseHandler;
 import com.ss.mqtt.broker.network.packet.PacketType;
 import com.ss.mqtt.broker.network.packet.in.handler.*;
 import com.ss.mqtt.broker.service.*;
@@ -20,7 +22,6 @@
 import org.jetbrains.annotations.NotNull;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
 import org.springframework.core.env.Environment;
 
 import java.net.InetSocketAddress;
@@ -48,14 +49,9 @@ private interface ChannelFactory extends
         return new DefaultBufferAllocator(networkConfig);
     }
 
-    @Bean
-    @NotNull ClientService clientService() {
-        return new DefaultClientService();
-    }
-
     @Bean
     @NotNull ClientIdRegistry clientIdRegistry() {
-        return new SimpleClientIdRegistry(
+        return new InMemoryClientIdRegistry(
             env.getProperty(
                 "client.id.available.chars",
                 "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-"
@@ -64,16 +60,20 @@ private interface ChannelFactory extends
         );
     }
 
+    @Bean
+    @NotNull MqttSessionService mqttSessionService() {
+        return new InMemoryMqttSessionService(
+            env.getProperty("sessions.clean.thread.interval", int.class, 60000)
+        );
+    }
+
     @Bean
     @NotNull CredentialSource credentialSource() {
         return new FileCredentialsSource(env.getProperty("credentials.source.file.name", "credentials"));
     }
 
     @Bean
-    @NotNull AuthenticationService authenticationService(
-        @NotNull CredentialSource credentialSource,
-        @NotNull ClientIdRegistry clientIdRegistry
-    ) {
+    @NotNull AuthenticationService authenticationService(@NotNull CredentialSource credentialSource) {
         return new SimpleAuthenticationService(
             credentialSource,
             env.getProperty("authentication.allow.anonymous", boolean.class, false)
@@ -85,31 +85,47 @@ private interface ChannelFactory extends
         @NotNull AuthenticationService authenticationService,
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull SubscriptionService subscriptionService,
-        @NotNull PublishingService publishingService
+        @NotNull PublishingService publishingService,
+        @NotNull MqttSessionService mqttSessionService
     ) {
 
         var handlers = new PacketInHandler[PacketType.INVALID.ordinal()];
-        handlers[PacketType.CONNECT.ordinal()] = new ConnectInPacketHandler(clientIdRegistry, authenticationService);
+        handlers[PacketType.CONNECT.ordinal()] = new ConnectInPacketHandler(
+            clientIdRegistry,
+            authenticationService,
+            mqttSessionService
+        );
         handlers[PacketType.SUBSCRIBE.ordinal()] = new SubscribeInPacketHandler(subscriptionService);
         handlers[PacketType.UNSUBSCRIBE.ordinal()] = new UnsubscribeInPacketHandler(subscriptionService);
         handlers[PacketType.PUBLISH.ordinal()] = new PublishInPacketHandler(publishingService);
+        handlers[PacketType.DISCONNECT.ordinal()] = new DisconnetInPacketHandler();
 
         return handlers;
     }
 
+    @Bean
+    @NotNull MqttClientReleaseHandler deviceMqttClientReleaseHandler(
+        @NotNull ClientIdRegistry clientIdRegistry,
+        @NotNull MqttSessionService mqttSessionService
+    ) {
+        return new DeviceMqttClientReleaseHandler(clientIdRegistry, mqttSessionService);
+    }
+
     @Bean
     @NotNull ServerNetwork<@NotNull MqttConnection> deviceNetwork(
         @NotNull ServerNetworkConfig networkConfig,
         @NotNull BufferAllocator bufferAllocator,
         @NotNull MqttConnectionConfig deviceConnectionConfig,
-        PacketInHandler @NotNull [] devicePacketHandlers
+        PacketInHandler @NotNull [] devicePacketHandlers,
+        @NotNull MqttClientReleaseHandler deviceMqttClientReleaseHandler
     ) {
         return NetworkFactory.newServerNetwork(
             networkConfig,
             deviceConnectionFactory(
                 bufferAllocator,
                 deviceConnectionConfig,
-                devicePacketHandlers
+                devicePacketHandlers,
+                deviceMqttClientReleaseHandler
             )
         );
     }
@@ -212,7 +228,8 @@ private interface ChannelFactory extends
     private @NotNull ChannelFactory deviceConnectionFactory(
         @NotNull BufferAllocator bufferAllocator,
         @NotNull MqttConnectionConfig connectionConfig,
-        PacketInHandler @NotNull [] packetHandlers
+        PacketInHandler @NotNull [] packetHandlers,
+        @NotNull MqttClientReleaseHandler deviceMqttClientReleaseHandler
     ) {
         return (network, channel) -> new MqttConnection(
             network,
@@ -221,7 +238,7 @@ private interface ChannelFactory extends
             100,
             packetHandlers,
             connectionConfig,
-            DeviceMqttClient::new
+            mqttConnection -> new DeviceMqttClient(mqttConnection, deviceMqttClientReleaseHandler)
         );
     }
 }

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

@@ -41,7 +41,7 @@ public interface MqttPropertyConstants {
 
     int SUBSCRIPTION_ID_NOT_DEFINED = 0;
 
-    boolean SESSIONS_ENABLED_DEFAULT = false;
+    boolean SESSIONS_ENABLED_DEFAULT = true;
     boolean KEEP_ALIVE_ENABLED_DEFAULT = false;
     boolean RETAIN_AVAILABLE_DEFAULT = false;
     boolean WILDCARD_SUBSCRIPTION_AVAILABLE_DEFAULT = false;

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

@@ -0,0 +1,18 @@
+package com.ss.mqtt.broker.model;
+
+import org.jetbrains.annotations.NotNull;
+
+public interface MqttSession {
+
+    interface UnsafeMqttSession extends MqttSession {
+
+        void setExpirationTime(long expirationTime);
+    }
+
+    @NotNull String getClientId();
+
+    /**
+     * @return the expiration time in ms or -1 if it should not be expired now.
+     */
+    long getExpirationTime();
+}

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

@@ -0,0 +1,22 @@
+package com.ss.mqtt.broker.model.impl;
+
+import com.ss.mqtt.broker.model.MqttSession.UnsafeMqttSession;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+import org.jetbrains.annotations.NotNull;
+
+@ToString
+@RequiredArgsConstructor
+public class DefaultMqttSession implements UnsafeMqttSession {
+
+    private final @NotNull String clientId;
+
+    private volatile @Getter @Setter long expirationTime = -1;
+
+    @Override
+    public @NotNull String getClientId() {
+        return clientId;
+    }
+}