finish the first part of improving subscription service

Author: JavaSaBr

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

@@ -272,9 +272,9 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
     return new MqttServerConnectionConfig(
         QoS.ofCode(env.getProperty("mqtt.connection.max.qos", int.class, 2)),
         env.getProperty(
-            "mqtt.external.connection.max.packet.size",
+            "mqtt.external.connection.max.message.size",
             int.class,
-            MqttProperties.MAXIMUM_PACKET_SIZE_DEFAULT),
+            MqttProperties.MAXIMUM_MESSAGE_SIZE_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.max.string.length",
             int.class,
@@ -294,11 +294,11 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
         env.getProperty(
             "mqtt.external.connection.receive.maximum",
             int.class,
-            MqttProperties.RECEIVE_MAXIMUM_DEFAULT),
+            MqttProperties.RECEIVE_MAXIMUM_PUBLISHES_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.topic.alias.maximum",
             int.class,
-            MqttProperties.TOPIC_ALIAS_MAXIMUM_DISABLED),
+            MqttProperties.TOPIC_ALIAS_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.default.session.expiration.time",
             long.class,

application/src/test/groovy/javasabr/mqtt/broker/application/IntegrationSpecification.groovy

@@ -142,7 +142,7 @@ class IntegrationSpecification extends Specification {
         MqttVersion.MQTT_5,
         MqttProperties.SESSION_EXPIRY_INTERVAL_DISABLED,
         serverConnConfig.receiveMaxPublishes(),
-        serverConnConfig.maxPacketSize(),
+        serverConnConfig.maxMessageSize(),
         serverConnConfig.topicAliasMaxValue(),
         MqttProperties.SERVER_KEEP_ALIVE_DEFAULT,
         false,
@@ -169,7 +169,7 @@ class IntegrationSpecification extends Specification {
         MqttVersion.MQTT_3_1_1,
         MqttProperties.SESSION_EXPIRY_INTERVAL_DISABLED,
         serverConnConfig.receiveMaxPublishes(),
-        serverConnConfig.maxPacketSize(),
+        serverConnConfig.maxMessageSize(),
         serverConnConfig.topicAliasMaxValue(),
         MqttProperties.SERVER_KEEP_ALIVE_DEFAULT,
         false,

model/src/main/java/javasabr/mqtt/model/MqttClientConnectionConfig.java

@@ -6,7 +6,7 @@ public record MqttClientConnectionConfig(
     MqttVersion mqttVersion,
     long sessionExpiryInterval,
     int receiveMaxPublishes,
-    int maxPacketSize,
+    int maxMessageSize,
     int topicAliasMaxValue,
     int keepAlive,
     boolean requestResponseInformation,

model/src/main/java/javasabr/mqtt/model/MqttMessageProperty.java

@@ -30,13 +30,13 @@ public enum MqttMessageProperty {
   RESPONSE_INFORMATION(0x1A, MqttDataType.UTF_8_STRING),
   SERVER_REFERENCE(0x1C, MqttDataType.UTF_8_STRING),
   REASON_STRING(0x1F, MqttDataType.UTF_8_STRING),
-  RECEIVE_MAXIMUM_PUBLISH(0x21, MqttDataType.SHORT),
+  RECEIVE_MAXIMUM_PUBLISHES(0x21, MqttDataType.SHORT),
   TOPIC_ALIAS_MAXIMUM(0x22, MqttDataType.SHORT),
   TOPIC_ALIAS(0x23, MqttDataType.SHORT),
   MAXIMUM_QOS(0x24, MqttDataType.BYTE),
   RETAIN_AVAILABLE(0x25, MqttDataType.BYTE),
   USER_PROPERTY(0x26, MqttDataType.UTF_8_STRING_PAIR),
-  MAXIMUM_PACKET_SIZE(0x27, MqttDataType.INTEGER),
+  MAXIMUM_MESSAGE_SIZE(0x27, MqttDataType.INTEGER),
   WILDCARD_SUBSCRIPTION_AVAILABLE(0x28, MqttDataType.BYTE),
   SUBSCRIPTION_IDENTIFIER_AVAILABLE(0x29, MqttDataType.BYTE),
   SHARED_SUBSCRIPTION_AVAILABLE(0x2A, MqttDataType.BYTE);

model/src/main/java/javasabr/mqtt/model/MqttProperties.java

@@ -4,7 +4,7 @@ public interface MqttProperties {
 
   QoS MAXIMUM_QOS_DEFAULT = QoS.EXACTLY_ONCE;
 
-  int MAXIMUM_PROTOCOL_PACKET_SIZE = 256 * 1024 * 1024;
+  int MAXIMUM_PROTOCOL_MESSAGE_SIZE = 256 * 1024 * 1024;
   int MAXIMUM_PACKET_ID = 0xFFFF;
 
   long SESSION_EXPIRY_INTERVAL_DISABLED = 0;
@@ -13,24 +13,20 @@ public interface MqttProperties {
   long SESSION_EXPIRY_INTERVAL_INFINITY = 0xFFFFFFFFL;
   long SESSION_EXPIRY_INTERVAL_UNDEFINED = -1;
 
-  int RECEIVE_MAXIMUM_UNDEFINED = -1;
-  int RECEIVE_MAXIMUM_MIN = 1;
-  int RECEIVE_MAXIMUM_DEFAULT = 10;
-  int RECEIVE_MAXIMUM_MAX = 0xFFFF;
+  int RECEIVE_MAXIMUM_PUBLISHES_UNDEFINED = -1;
+  int RECEIVE_MAXIMUM_PUBLISHES_MIN = 1;
+  int RECEIVE_MAXIMUM_PUBLISHES_DEFAULT = 10;
+  int RECEIVE_MAXIMUM_PUBLISHES_MAX = 0xFFFF;
 
-  int MAXIMUM_PACKET_SIZE_UNDEFINED = -1;
-  int MAXIMUM_PACKET_SIZE_DEFAULT = 1024;
-  int MAXIMUM_PACKET_SIZE_MIN = 1;
-  int MAXIMUM_PACKET_SIZE_MAX = MAXIMUM_PROTOCOL_PACKET_SIZE;
+  int MAXIMUM_MESSAGE_SIZE_UNDEFINED = -1;
+  int MAXIMUM_MESSAGE_SIZE_DEFAULT = 3074;
+  int MAXIMUM_MESSAGE_SIZE_MIN = 128;
+  int MAXIMUM_MESSAGE_SIZE_MAX = MAXIMUM_PROTOCOL_MESSAGE_SIZE;
 
-  int MAXIMUM_STRING_LENGTH = 2048;
+  int MAXIMUM_STRING_LENGTH = 1024;
   int MAXIMUM_BINARY_SIZE = 2048;
   int MAXIMUM_TOPIC_LEVELS = 10;
 
-  int PAYLOAD_FORMAT_UNDEFINED = -1;
-  int PAYLOAD_FORMAT_BINARY = 0;
-  int PAYLOAD_FORMAT_UTF8_STRING = 1;
-
   long MESSAGE_EXPIRY_INTERVAL_UNDEFINED = -1;
   long MESSAGE_EXPIRY_INTERVAL_INFINITY = 0;
 
@@ -45,6 +41,7 @@ public interface MqttProperties {
 
   int TOPIC_ALIAS_UNDEFINED = 0;
   int TOPIC_ALIAS_MIN = 1;
+  int TOPIC_ALIAS_DEFAULT = 10;
   int TOPIC_ALIAS_MAX = 0xFFFF;
   int TOPIC_ALIAS_NOT_SET = 0;
 
@@ -58,7 +55,7 @@ public interface MqttProperties {
   boolean RETAIN_AVAILABLE_DEFAULT = false;
   boolean WILDCARD_SUBSCRIPTION_AVAILABLE_DEFAULT = false;
   boolean SHARED_SUBSCRIPTION_AVAILABLE_DEFAULT = false;
-  boolean SUBSCRIPTION_IDENTIFIER_AVAILABLE_DEFAULT = false;
+  boolean SUBSCRIPTION_IDENTIFIER_AVAILABLE_DEFAULT = true;
 
   int PACKET_ID_FOR_QOS_0 = 0;
 }

model/src/main/java/javasabr/mqtt/model/MqttServerConnectionConfig.java

@@ -1,8 +1,10 @@
 package javasabr.mqtt.model;
 
+import javasabr.rlib.common.util.NumberUtils;
+
 public record MqttServerConnectionConfig(
     QoS maxQos,
-    int maxPacketSize,
+    int maxMessageSize,
     int maxStringLength,
     int maxBinarySize,
     int maxTopicLevels,
@@ -16,4 +18,112 @@ public record MqttServerConnectionConfig(
     boolean wildcardSubscriptionAvailable,
     boolean subscriptionIdAvailable,
     boolean sharedSubscriptionAvailable) {
+
+  public MqttServerConnectionConfig(
+      QoS maxQos,
+      int maxMessageSize,
+      int maxStringLength,
+      int maxBinarySize,
+      int maxTopicLevels,
+      int minKeepAliveTime,
+      int receiveMaxPublishes,
+      int topicAliasMaxValue,
+      long defaultSessionExpiryInterval,
+      boolean keepAliveEnabled,
+      boolean sessionsEnabled,
+      boolean retainAvailable,
+      boolean wildcardSubscriptionAvailable,
+      boolean subscriptionIdAvailable,
+      boolean sharedSubscriptionAvailable) {
+    this.maxQos = maxQos;
+    this.maxMessageSize = NumberUtils.validate(
+        maxMessageSize,
+        MqttProperties.MAXIMUM_MESSAGE_SIZE_MIN,
+        MqttProperties.MAXIMUM_MESSAGE_SIZE_MAX);
+    this.maxStringLength = NumberUtils.validate(
+        maxStringLength,
+        1,
+        maxMessageSize / 2);
+    this.maxBinarySize = NumberUtils.validate(
+        maxBinarySize,
+        1,
+        maxMessageSize);
+    this.maxTopicLevels = NumberUtils.validate(
+        maxTopicLevels,
+        1,
+        Byte.MAX_VALUE);
+    this.minKeepAliveTime = NumberUtils.validate(
+        minKeepAliveTime,
+        MqttProperties.SERVER_KEEP_ALIVE_MIN,
+        MqttProperties.SERVER_KEEP_ALIVE_MAX);
+    this.receiveMaxPublishes = receiveMaxPublishes;
+    this.topicAliasMaxValue = NumberUtils.validate(
+        topicAliasMaxValue,
+        MqttProperties.TOPIC_ALIAS_MIN,
+        MqttProperties.TOPIC_ALIAS_MAX);
+    this.defaultSessionExpiryInterval = defaultSessionExpiryInterval;
+    this.keepAliveEnabled = keepAliveEnabled;
+    this.sessionsEnabled = sessionsEnabled;
+    this.retainAvailable = retainAvailable;
+    this.wildcardSubscriptionAvailable = wildcardSubscriptionAvailable;
+    this.subscriptionIdAvailable = subscriptionIdAvailable;
+    this.sharedSubscriptionAvailable = sharedSubscriptionAvailable;
+  }
+
+  public MqttServerConnectionConfig withMaxQos(QoS maxQos) {
+    return new MqttServerConnectionConfig(
+        maxQos,
+        maxMessageSize,
+        maxStringLength,
+        maxBinarySize,
+        maxTopicLevels,
+        minKeepAliveTime,
+        receiveMaxPublishes,
+        topicAliasMaxValue,
+        defaultSessionExpiryInterval,
+        keepAliveEnabled,
+        sessionsEnabled,
+        retainAvailable,
+        wildcardSubscriptionAvailable,
+        subscriptionIdAvailable,
+        sharedSubscriptionAvailable);
+  }
+
+  public MqttServerConnectionConfig withWildcardSubscriptionAvailable(boolean wildcardSubscriptionAvailable) {
+    return new MqttServerConnectionConfig(
+        maxQos,
+        maxMessageSize,
+        maxStringLength,
+        maxBinarySize,
+        maxTopicLevels,
+        minKeepAliveTime,
+        receiveMaxPublishes,
+        topicAliasMaxValue,
+        defaultSessionExpiryInterval,
+        keepAliveEnabled,
+        sessionsEnabled,
+        retainAvailable,
+        wildcardSubscriptionAvailable,
+        subscriptionIdAvailable,
+        sharedSubscriptionAvailable);
+  }
+
+  public MqttServerConnectionConfig withSharedSubscriptionAvailable(boolean sharedSubscriptionAvailable) {
+    return new MqttServerConnectionConfig(
+        maxQos,
+        maxMessageSize,
+        maxStringLength,
+        maxBinarySize,
+        maxTopicLevels,
+        minKeepAliveTime,
+        receiveMaxPublishes,
+        topicAliasMaxValue,
+        defaultSessionExpiryInterval,
+        keepAliveEnabled,
+        sessionsEnabled,
+        retainAvailable,
+        wildcardSubscriptionAvailable,
+        subscriptionIdAvailable,
+        sharedSubscriptionAvailable);
+  }
 }

model/src/main/java/javasabr/mqtt/model/subscribtion/Subscription.java

@@ -1,6 +1,7 @@
 package javasabr.mqtt.model.subscribtion;
 
 import com.fasterxml.jackson.annotation.JsonValue;
+import javasabr.mqtt.model.MqttProperties;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.SubscribeRetainHandling;
 import javasabr.mqtt.model.topic.TopicFilter;
@@ -10,6 +11,10 @@ public record Subscription(
       The subscriber's topic filter.
      */
     TopicFilter topicFilter,
+    /*
+     * The associated ID for the subscription
+     */
+    int subscriptionId,
     /*
       Maximum QoS field. This gives the maximum QoS level at which the Server can send Application Messages to the
       Client.
@@ -36,6 +41,7 @@ public record Subscription(
   public static Subscription minimal(TopicFilter topicFilter, QoS qos) {
     return new Subscription(
         topicFilter,
+        MqttProperties.SUBSCRIPTION_ID_UNDEFINED,
         qos,
         SubscribeRetainHandling.SEND,
         true,

model/src/test/groovy/javasabr/mqtt/model/topic/tree/TopicTreeTest.groovy

@@ -1,5 +1,6 @@
 package javasabr.mqtt.model.topic.tree
 
+import javasabr.mqtt.model.MqttProperties
 import javasabr.mqtt.model.QoS
 import javasabr.mqtt.model.SubscribeRetainHandling
 import javasabr.mqtt.model.subscriber.SingleSubscriber
@@ -545,6 +546,7 @@ class TopicTreeTest extends UnitSpecification {
   static def makeSubscription(String topicFilter) {
     return new Subscription(
         TopicFilter.valueOf(topicFilter),
+        MqttProperties.SUBSCRIPTION_ID_UNDEFINED,
         QoS.AT_LEAST_ONCE,
         SubscribeRetainHandling.SEND,
         true,
@@ -554,6 +556,7 @@ class TopicTreeTest extends UnitSpecification {
   static def makeSharedSubscription(String topicFilter) {
     return new Subscription(
         SharedTopicFilter.valueOf(topicFilter),
+        MqttProperties.SUBSCRIPTION_ID_UNDEFINED,
         QoS.AT_LEAST_ONCE,
         SubscribeRetainHandling.SEND,
         true,
@@ -563,6 +566,7 @@ class TopicTreeTest extends UnitSpecification {
   static def makeSubscription(String topicFilter, int qos) {
     return new Subscription(
         TopicFilter.valueOf(topicFilter),
+        MqttProperties.SUBSCRIPTION_ID_UNDEFINED,
         QoS.ofCode(qos),
         SubscribeRetainHandling.SEND,
         true,

network/src/main/java/javasabr/mqtt/network/MqttSession.java

@@ -61,4 +61,6 @@ default void resend(MqttClient client, Publish publish) {}
   void removeSubscription(TopicFilter subscribe);
 
   Array<Subscription> storedSubscriptions();
+
+  Array<Subscription> findStoredSubscriptionWithId(int subscriptionId);
 }

network/src/main/java/javasabr/mqtt/network/message/in/ConnectAckMqttInMessage.java

@@ -49,7 +49,7 @@ public class ConnectAckMqttInMessage extends MqttInMessage {
 
         If the Receive Maximum value is absent, then its value defaults to 65,535.
        */
-      MqttMessageProperty.RECEIVE_MAXIMUM_PUBLISH,
+      MqttMessageProperty.RECEIVE_MAXIMUM_PUBLISHES,
       /*
         Followed by a Byte with a value of either 0 or 1. It is a Protocol Error to include Maximum QoS more than
         once, or to have a value other than 0 or 1. If the Maximum QoS is absent, the Client uses a Maximum
@@ -104,7 +104,7 @@ public class ConnectAckMqttInMessage extends MqttInMessage {
         a Server receives a packet whose size exceeds this limit, this is a Protocol Error, the Server uses
         DISCONNECT with Reason Code 0x95 (Packet too large), as described in section 4.13.
        */
-      MqttMessageProperty.MAXIMUM_PACKET_SIZE,
+      MqttMessageProperty.MAXIMUM_MESSAGE_SIZE,
       /*
         Followed by the UTF-8 string which is the Assigned Client Identifier. It is a Protocol Error to include the
         Assigned Client Identifier more than once.
@@ -261,7 +261,7 @@ public class ConnectAckMqttInMessage extends MqttInMessage {
   long sessionExpiryInterval;
 
   int receiveMaxPublishes;
-  int maxPacketSize;
+  int maxMessageSize;
   int topicAliasMaxValue;
   int serverKeepAlive;
 
@@ -286,10 +286,10 @@ public ConnectAckMqttInMessage(byte info) {
     this.authenticationMethod = StringUtils.EMPTY;
     this.authenticationData = ArrayUtils.EMPTY_BYTE_ARRAY;
     this.serverKeepAlive = MqttProperties.SERVER_KEEP_ALIVE_UNDEFINED;
-    this.maxPacketSize = MqttProperties.MAXIMUM_PACKET_SIZE_UNDEFINED;
+    this.maxMessageSize = MqttProperties.MAXIMUM_MESSAGE_SIZE_UNDEFINED;
     this.sessionExpiryInterval = MqttProperties.SESSION_EXPIRY_INTERVAL_UNDEFINED;
     this.topicAliasMaxValue = MqttProperties.TOPIC_ALIAS_MAXIMUM_UNDEFINED;
-    this.receiveMaxPublishes = MqttProperties.RECEIVE_MAXIMUM_UNDEFINED;
+    this.receiveMaxPublishes = MqttProperties.RECEIVE_MAXIMUM_PUBLISHES_UNDEFINED;
   }
 
   @Override
@@ -336,10 +336,10 @@ protected void applyProperty(MqttMessageProperty property, long value) {
       case SHARED_SUBSCRIPTION_AVAILABLE -> sharedSubscriptionAvailable = NumberUtils.toBoolean(value);
       case SUBSCRIPTION_IDENTIFIER_AVAILABLE -> subscriptionIdAvailable = NumberUtils.toBoolean(value);
       case RETAIN_AVAILABLE -> retainAvailable = NumberUtils.toBoolean(value);
-      case RECEIVE_MAXIMUM_PUBLISH -> receiveMaxPublishes = (int) NumberUtils.validate(
+      case RECEIVE_MAXIMUM_PUBLISHES -> receiveMaxPublishes = (int) NumberUtils.validate(
           value,
-          MqttProperties.RECEIVE_MAXIMUM_MIN,
-          MqttProperties.RECEIVE_MAXIMUM_MAX);
+          MqttProperties.RECEIVE_MAXIMUM_PUBLISHES_MIN,
+          MqttProperties.RECEIVE_MAXIMUM_PUBLISHES_MAX);
       case MAXIMUM_QOS -> maximumQos = QoS.ofCode((int) value);
       case SERVER_KEEP_ALIVE -> serverKeepAlive = NumberUtils.validate(
           (int) value,
@@ -353,10 +353,10 @@ protected void applyProperty(MqttMessageProperty property, long value) {
           value,
           MqttProperties.SESSION_EXPIRY_INTERVAL_MIN,
           MqttProperties.SESSION_EXPIRY_INTERVAL_INFINITY);
-      case MAXIMUM_PACKET_SIZE -> maxPacketSize = NumberUtils.validate(
+      case MAXIMUM_MESSAGE_SIZE -> maxMessageSize = NumberUtils.validate(
           (int) value,
-          MqttProperties.MAXIMUM_PACKET_SIZE_MIN,
-          MqttProperties.MAXIMUM_PACKET_SIZE_MAX);
+          MqttProperties.MAXIMUM_MESSAGE_SIZE_MIN,
+          MqttProperties.MAXIMUM_MESSAGE_SIZE_MAX);
       default -> unexpectedProperty(property);
     }
   }