Improve Publish Implementation, part 1 (#53)

Implement full validation of incoming publish messages

Author: JavaSaBr

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

@@ -9,6 +9,7 @@
 import javasabr.mqtt.network.MqttConnection;
 import javasabr.mqtt.network.MqttConnectionFactory;
 import javasabr.mqtt.network.handler.MqttClientReleaseHandler;
+import javasabr.mqtt.network.impl.ExternalMqttClient;
 import javasabr.mqtt.service.AuthenticationService;
 import javasabr.mqtt.service.ClientIdRegistry;
 import javasabr.mqtt.service.ConnectionService;
@@ -193,8 +194,8 @@ MqttInMessageHandler unsubscribeMqttInMessageHandler(
   }
 
   @Bean
-  ConnectionService mqttConnectionService(Collection<? extends MqttInMessageHandler> inMessageHandlers) {
-    return new DefaultConnectionService(inMessageHandlers);
+  ConnectionService externalMqttConnectionService(Collection<? extends MqttInMessageHandler> inMessageHandlers) {
+    return new DefaultConnectionService(ExternalMqttClient.class, inMessageHandlers);
   }
 
   @Bean
@@ -298,7 +299,7 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
         env.getProperty(
             "mqtt.external.connection.topic.alias.maximum",
             int.class,
-            MqttProperties.TOPIC_ALIAS_DEFAULT),
+            MqttProperties.TOPIC_ALIAS_MAX_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.default.session.expiration.time",
             long.class,

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

@@ -152,6 +152,7 @@ class IntegrationSpecification extends Specification {
       isSupported(MqttVersion.MQTT_5) >> true
       isSupported(MqttVersion.MQTT_3_1_1) >> true
       serverConnectionConfig() >> serverConnConfig
+      clientConnectionConfig() >> clientConnConfig
       client() >> Stub(UnsafeMqttClient) {
         connectionConfig() >> clientConnConfig
         connection() >> connectionRef.get()
@@ -179,6 +180,7 @@ class IntegrationSpecification extends Specification {
       isSupported(MqttVersion.MQTT_5) >> false
       isSupported(MqttVersion.MQTT_3_1_1) >> true
       serverConnectionConfig() >> serverConnConfig
+      clientConnectionConfig() >> clientConnConfig
       client() >> Stub(UnsafeMqttClient) {
         connectionConfig() >> clientConnConfig
         connection() >> connectionRef.get()

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

@@ -40,14 +40,15 @@ class PublishRetryTest extends IntegrationSpecification {
         subscriber.send(new ConnectMqtt311OutMessage(subscriberId, keepAlive))
         def connectAck = subscriber.readNext() as ConnectAckMqttInMessage
     then:
-        connectAck.reasonCode == ConnectAckReasonCode.SUCCESS
+        connectAck.reasonCode() == ConnectAckReasonCode.SUCCESS
     when:
         subscriber.send(new SubscribeMqtt311OutMessage(
             1,
             Array.of(Subscription.minimal(TopicFilter.valueOf("test/retry/$subscriberId"), QoS.AT_LEAST_ONCE))))
         def subscribeAck = subscriber.readNext() as SubscribeAckMqttInMessage
     then:
-        subscribeAck.reasonCodes.stream()
+        subscribeAck.reasonCodes()
+            .stream()
             .allMatch({ it == SubscribeAckReasonCode.GRANTED_QOS_1 })
     when:
         publisher.publishWith()
@@ -58,7 +59,7 @@ class PublishRetryTest extends IntegrationSpecification {
             .join()
         def receivedPublish = subscriber.readNext() as PublishMqttInMessage
     then:
-        receivedPublish.payload == publishPayload
+        receivedPublish.payload() == publishPayload
     when:
         subscriber.disconnect()
         Thread.sleep(1_000)

gradle/libs.versions.toml

@@ -32,6 +32,7 @@ moquette-broker = "0.17"
 # https://mvnrepository.com/artifact/tools.jackson.core/jackson-core
 jackson = "3.0.1"
 
+
 [libraries]
 rlib-network = { module = "javasabr.rlib:rlib-network", version.ref = "rlib" }
 rlib-logger-api = { module = "javasabr.rlib:rlib-logger-api", version.ref = "rlib" }

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

@@ -39,4 +39,8 @@ public int maxTopicLevels() {
   public int maxStringLength() {
     return server.maxStringLength();
   }
+
+  public int maxBinarySize() {
+    return server.maxBinarySize();
+  }
 }

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

@@ -27,8 +27,9 @@ public interface MqttProperties {
   int MAXIMUM_BINARY_SIZE = 2048;
   int MAXIMUM_TOPIC_LEVELS = 10;
 
-  long MESSAGE_EXPIRY_INTERVAL_UNDEFINED = -1;
+  long MESSAGE_EXPIRY_INTERVAL_IS_NOT_SET = -1;
   long MESSAGE_EXPIRY_INTERVAL_INFINITY = 0;
+  long MESSAGE_EXPIRY_INTERVAL_MIN = 0;
 
   int TOPIC_ALIAS_MAXIMUM_UNDEFINED = -1;
   int TOPIC_ALIAS_MAXIMUM_DISABLED = 0;
@@ -39,11 +40,11 @@ public interface MqttProperties {
   int SERVER_KEEP_ALIVE_MIN = 0;
   int SERVER_KEEP_ALIVE_MAX = 0xFFFF;
 
-  int TOPIC_ALIAS_UNDEFINED = 0;
   int TOPIC_ALIAS_MIN = 1;
-  int TOPIC_ALIAS_DEFAULT = 10;
+  int TOPIC_ALIAS_MAX_DEFAULT = 10;
   int TOPIC_ALIAS_MAX = 0xFFFF;
-  int TOPIC_ALIAS_NOT_SET = 0;
+  int TOPIC_ALIAS_NOT_SET = Integer.MIN_VALUE;
+  int TOPIC_ALIAS_INVALID = 0;
 
   int SUBSCRIPTION_ID_IS_NOT_SET = 0;
   int MESSAGE_ID_IS_NOT_SET = 0;

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

@@ -1,6 +1,15 @@
 package javasabr.mqtt.model;
 
 public interface MqttProtocolErrors {
-  String NO_ANY_TOPIC_FILTER = "No any topic filters";
-  String UNSUPPORTED_QOS_OR_RETAIN_HANDLING = "Unsupported qos or retain handling";
+  String NO_ANY_TOPIC_FILTER = "Not provided any information about TopicFilters";
+  String NO_ANY_TOPIC_NANE = "Not provided any information about TopicName";
+  //String INVALID_TOPIC_ALIAS = "Provided invalid TopicAlias";
+  String INVALID_PAYLOAD_FORMAT = "Provided invalid PayloadFormat";
+  String INVALID_MESSAGE_EXPIRY_INTERVAL = "Provided invalid MessageExpiryInterval";
+  String INVALID_RESPONSE_TOPIC_NAME = "Provided invalid ResponseTopicName";
+  String UNSUPPORTED_QOS_OR_RETAIN_HANDLING = "Unsupported 'QoS' or 'RetainHandling'";
+  String MISSED_REQUIRED_MESSAGE_ID = "'Packet Identifier' must be presented'";
+  String PROTOCOL_LEVEL_UNSUPPORTED_NO_LOCAL_OPTION = "'NoLocal' option is not available on this protocol level";
+  String PROTOCOL_LEVEL_UNSUPPORTED_RETAIN_AS_PUBLISH_OPTION = "'RetainAsPublished' option is not available on this protocol level";
+  String PROTOCOL_LEVEL_UNSUPPORTED_RETAIN_HANDLING_OPTION = "'RetainHandling' option is not available on this protocol level";
 }

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

@@ -1,7 +1,9 @@
 package javasabr.mqtt.model;
 
 import javasabr.rlib.common.util.NumberUtils;
+import lombok.Builder;
 
+@Builder(toBuilder = true)
 public record MqttServerConnectionConfig(
     QoS maxQos,
     int maxMessageSize,
@@ -71,78 +73,32 @@ public MqttServerConnectionConfig(
   }
 
   public MqttServerConnectionConfig withMaxQos(QoS maxQos) {
-    return new MqttServerConnectionConfig(
-        maxQos,
-        maxMessageSize,
-        maxStringLength,
-        maxBinarySize,
-        maxTopicLevels,
-        minKeepAliveTime,
-        receiveMaxPublishes,
-        topicAliasMaxValue,
-        defaultSessionExpiryInterval,
-        keepAliveEnabled,
-        sessionsEnabled,
-        retainAvailable,
-        wildcardSubscriptionAvailable,
-        subscriptionIdAvailable,
-        sharedSubscriptionAvailable);
+    return toBuilder()
+        .maxQos(maxQos)
+        .build();
   }
 
   public MqttServerConnectionConfig withWildcardSubscriptionAvailable(boolean wildcardSubscriptionAvailable) {
-    return new MqttServerConnectionConfig(
-        maxQos,
-        maxMessageSize,
-        maxStringLength,
-        maxBinarySize,
-        maxTopicLevels,
-        minKeepAliveTime,
-        receiveMaxPublishes,
-        topicAliasMaxValue,
-        defaultSessionExpiryInterval,
-        keepAliveEnabled,
-        sessionsEnabled,
-        retainAvailable,
-        wildcardSubscriptionAvailable,
-        subscriptionIdAvailable,
-        sharedSubscriptionAvailable);
+    return toBuilder()
+        .wildcardSubscriptionAvailable(wildcardSubscriptionAvailable)
+        .build();
   }
 
   public MqttServerConnectionConfig withSharedSubscriptionAvailable(boolean sharedSubscriptionAvailable) {
-    return new MqttServerConnectionConfig(
-        maxQos,
-        maxMessageSize,
-        maxStringLength,
-        maxBinarySize,
-        maxTopicLevels,
-        minKeepAliveTime,
-        receiveMaxPublishes,
-        topicAliasMaxValue,
-        defaultSessionExpiryInterval,
-        keepAliveEnabled,
-        sessionsEnabled,
-        retainAvailable,
-        wildcardSubscriptionAvailable,
-        subscriptionIdAvailable,
-        sharedSubscriptionAvailable);
+    return toBuilder()
+        .sharedSubscriptionAvailable(sharedSubscriptionAvailable)
+        .build();
   }
 
   public MqttServerConnectionConfig withSubscriptionIdAvailable(boolean subscriptionIdAvailable) {
-    return new MqttServerConnectionConfig(
-        maxQos,
-        maxMessageSize,
-        maxStringLength,
-        maxBinarySize,
-        maxTopicLevels,
-        minKeepAliveTime,
-        receiveMaxPublishes,
-        topicAliasMaxValue,
-        defaultSessionExpiryInterval,
-        keepAliveEnabled,
-        sessionsEnabled,
-        retainAvailable,
-        wildcardSubscriptionAvailable,
-        subscriptionIdAvailable,
-        sharedSubscriptionAvailable);
+    return toBuilder()
+        .subscriptionIdAvailable(subscriptionIdAvailable)
+        .build();
+  }
+
+  public MqttServerConnectionConfig withRetainAvailable(boolean retainAvailable) {
+    return toBuilder()
+        .retainAvailable(retainAvailable)
+        .build();
   }
 }

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

@@ -37,4 +37,8 @@ public int number() {
   public QoS lower(QoS alternative) {
     return level > alternative.level ? alternative : this;
   }
+
+  public boolean isLower(QoS another) {
+    return level < another.level;
+  }
 }

model/src/main/java/javasabr/mqtt/model/topic/AbstractTopic.java

@@ -1,5 +1,6 @@
 package javasabr.mqtt.model.topic;
 
+import java.util.Objects;
 import javasabr.mqtt.base.util.DebugUtils;
 import javasabr.rlib.common.util.StringUtils;
 import lombok.AccessLevel;
@@ -53,6 +54,20 @@ public String toString() {
     return rawTopic;
   }
 
+  @Override
+  public boolean equals(Object object) {
+    if (object == null || getClass() != object.getClass()) {
+      return false;
+    }
+    var that = (AbstractTopic) object;
+    return Objects.equals(rawTopic, that.rawTopic);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(rawTopic);
+  }
+
   protected static String[] splitTopic(String topic) {
     int segmentCount = countOccurrencesOf(topic, AbstractTopic.DELIMITER) + 1;
     var segments = new String[segmentCount];