Improve publish implementation, part 2 (#56)

Improve publish implementation, part 2 (#56)

Author: JavaSaBr

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

@@ -228,8 +228,12 @@ PublishDeliveringService publishDeliveringService(
   @Bean
   MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
-      PublishDeliveringService publishDeliveringService) {
-    return new Qos0MqttPublishInMessageHandler(subscriptionService, publishDeliveringService);
+      PublishDeliveringService publishDeliveringService,
+      MessageOutFactoryService messageOutFactoryService) {
+    return new Qos0MqttPublishInMessageHandler(
+        subscriptionService,
+        publishDeliveringService,
+        messageOutFactoryService);
   }
 
   @Bean

lombok.config

@@ -1 +1,3 @@
-lombok.log.custom.declaration = javasabr.rlib.logger.api.Logger javasabr.rlib.logger.api.LoggerManager.getLogger(TYPE)
+lombok.log.custom.declaration = javasabr.rlib.logger.api.Logger javasabr.rlib.logger.api.LoggerManager.getLogger(TYPE)
+lombok.accessors.fluent=true
+lombok.accessors.chain=false

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

@@ -9,6 +9,7 @@ public interface MqttProtocolErrors {
   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 NOT_EXPECTED_MESSAGE_ID = "'Packet Identifier' must be zero'";
   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/MqttUser.java

@@ -0,0 +1,3 @@
+package javasabr.mqtt.model;
+
+public interface MqttUser {}

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

@@ -10,8 +10,8 @@
 import lombok.experimental.FieldDefaults;
 
 @Getter
+@Accessors
 @RequiredArgsConstructor
-@Accessors(fluent = true, chain = false)
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public enum QoS implements NumberedEnum<QoS> {
   AT_MOST_ONCE(0, SubscribeAckReasonCode.GRANTED_QOS_0),
@@ -38,7 +38,11 @@ public QoS lower(QoS alternative) {
     return level > alternative.level ? alternative : this;
   }
 
-  public boolean isLower(QoS another) {
+  public boolean isLowerThan(QoS another) {
     return level < another.level;
   }
+
+  public boolean isHigherThan(QoS another) {
+    return level > another.level;
+  }
 }

network/src/main/java/javasabr/mqtt/network/message/MqttMessageType.java renamed to model/src/main/java/javasabr/mqtt/model/message/MqttMessageType.java

@@ -1,4 +1,4 @@
-package javasabr.mqtt.network.message;
+package javasabr.mqtt.model.message;
 
 import lombok.AccessLevel;
 import lombok.CustomLog;
@@ -41,12 +41,12 @@ public enum MqttMessageType {
   /**
    * A PUBREL Packet is the response to a PUBREC Packet. It is the third packet of the QoS 2 protocol exchange.
    */
-  PUBLISH_RELEASED(6),
+  PUBLISH_RELEASE(6),
   /**
    * The PUBCOMP packet is the response to a PUBREL packet. It is the fourth and final packet of the QoS 2 protocol
    * exchange.
    */
-  PUBLISH_COMPLETED(7),
+  PUBLISH_COMPLETE(7),
   /**
    * The SUBSCRIBE Packet is sent from the Client to the Server to create one or more Subscriptions. Each Subscription
    * registers a Client’s interest in one or more Topics. The Server sends PUBLISH Packets to the Client in order to
@@ -98,7 +98,7 @@ public enum MqttMessageType {
    * such as challenge / response authentication. It is a Protocol Error for the Client or Server to send an AUTH packet
    * if the CONNECT packet did not contain the same Authentication Method.
    */
-  AUTHENTICATE(15),
+  AUTHENTICATION(15),
 
   /**
    * Not supported
@@ -108,7 +108,7 @@ public enum MqttMessageType {
   private static final MqttMessageType[] VALUES = values();
 
   public static MqttMessageType fromByte(byte messageType) {
-    if (messageType < 0 || messageType > AUTHENTICATE.typeIndex()) {
+    if (messageType < 0 || messageType > AUTHENTICATION.typeIndex()) {
       log.warning(messageType, "Invalid message type:[%s]"::formatted);
       return INVALID;
     } else {

model/src/main/java/javasabr/mqtt/model/publishing/Publish.java

@@ -1,6 +1,7 @@
 package javasabr.mqtt.model.publishing;
 
 import javasabr.mqtt.base.util.DebugUtils;
+import javasabr.mqtt.model.MqttProperties;
 import javasabr.mqtt.model.PayloadFormat;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.TrackableMessage;
@@ -26,10 +27,34 @@ public record Publish(
     PayloadFormat payloadFormat,
     Array<StringPair> userProperties) implements TrackableMessage {
 
+  private static final Array<StringPair> EMPTY_USER_PROPERTIES = Array.empty(StringPair.class);
+
   static {
     DebugUtils.registerIncludedFields("topicName", "messageId", "qos", "topicAlias", "payloadFormat");
   }
 
+  public static Publish minimal(int messageId, QoS qos, TopicName topicName, byte[] payload) {
+    return new Publish(
+        messageId,
+        qos,
+        topicName,
+        null,
+        payload,
+        false,
+        false,
+        null,
+        IntArray.EMPTY,
+        null,
+        MqttProperties.MESSAGE_EXPIRY_INTERVAL_IS_NOT_SET,
+        MqttProperties.TOPIC_ALIAS_NOT_SET,
+        PayloadFormat.BINARY,
+        EMPTY_USER_PROPERTIES);
+  }
+
+  public static Publish minimal(QoS qos, TopicName topicName, byte[] payload) {
+    return minimal(MqttProperties.MESSAGE_ID_IS_NOT_SET, qos, topicName, payload);
+  }
+
   public Publish withDuplicated() {
     if (duplicated) {
       return this;

model/src/main/java/javasabr/mqtt/model/reason/code/AuthenticateReasonCode.java

@@ -27,7 +27,7 @@ public enum AuthenticateReasonCode {
 
     var maxId = Stream
         .of(values())
-        .mapToInt(AuthenticateReasonCode::getValue)
+        .mapToInt(AuthenticateReasonCode::value)
         .max()
         .orElse(0);
 

model/src/main/java/javasabr/mqtt/model/reason/code/ConnectAckReasonCode.java

@@ -105,7 +105,7 @@ public enum ConnectAckReasonCode {
 
     var maxId = Stream
         .of(values())
-        .mapToInt(ConnectAckReasonCode::getMqtt5)
+        .mapToInt(ConnectAckReasonCode::mqtt5)
         .map(value -> Byte.toUnsignedInt((byte) value))
         .max()
         .orElse(0);

model/src/main/java/javasabr/mqtt/model/reason/code/PublishAckReasonCode.java

@@ -1,87 +1,69 @@
 package javasabr.mqtt.model.reason.code;
 
-import java.util.stream.Stream;
-import javasabr.rlib.common.util.ObjectUtils;
-import lombok.AccessLevel;
+import javasabr.rlib.common.util.NumberedEnum;
+import javasabr.rlib.common.util.NumberedEnumMap;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.Accessors;
-import lombok.experimental.FieldDefaults;
 
 @Getter
+@Accessors
 @RequiredArgsConstructor
-@Accessors(fluent = true, chain = false)
-@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
-public enum PublishAckReasonCode {
+public enum PublishAckReasonCode implements NumberedEnum<PublishAckReasonCode>, ReasonCode {
   /**
    * The message is accepted. Publication of the QoS 1 message proceeds.
    */
-  SUCCESS((byte) 0x00),
+  SUCCESS(0x00),
   /**
    * The message is accepted but there are no subscribers. This is sent only by the Server. If the Server knows that
    * there are no matching subscribers, it MAY use this Reason Code instead of 0x00 (Success).
    */
-  NO_MATCHING_SUBSCRIBERS((byte) 0x10),
+  NO_MATCHING_SUBSCRIBERS(0x10),
 
   // ERRORS
 
   /**
    * The receiver does not accept the publish but either does not want to reveal the reason, or it does not match one of
    * the other values.
    */
-  UNSPECIFIED_ERROR((byte) 0x80),
+  UNSPECIFIED_ERROR(0x80),
   /**
    * The PUBLISH is valid but the receiver is not willing to accept it.
    */
-  IMPLEMENTATION_SPECIFIC_ERROR((byte) 0x83),
+  IMPLEMENTATION_SPECIFIC_ERROR(0x83),
   /**
    * The PUBLISH is not authorized.
    */
-  NOT_AUTHORIZED((byte) 0x87),
+  NOT_AUTHORIZED(0x87),
   /**
    * The Topic Name is not malformed, but is not accepted by this Client or Server.
    */
-  TOPIC_NAME_INVALID((byte) 0x90),
+  TOPIC_NAME_INVALID(0x90),
   /**
    * The Packet Identifier is already in use. This might indicate a mismatch in the Session State between the Client and
    * Server.
    */
-  PACKET_IDENTIFIER_IN_USE((byte) 0x91),
+  PACKET_IDENTIFIER_IN_USE(0x91),
   /**
    * An implementation or administrative imposed limit has been exceeded.
    */
-  QUOTA_EXCEEDED((byte) 0x97),
+  QUOTA_EXCEEDED(0x97),
   /**
    * The payload format does not match the specified Payload Format Indicator.
    */
-  PAYLOAD_FORMAT_INVALID((byte) 0x99);
+  PAYLOAD_FORMAT_INVALID(0x99);
 
-  private static final PublishAckReasonCode[] VALUES;
+  private static final NumberedEnumMap<PublishAckReasonCode> NUMBERED_MAP =
+      new NumberedEnumMap<>(PublishAckReasonCode.class);
 
-  static {
-
-    int maxValue = Stream
-        .of(values())
-        .mapToInt(PublishAckReasonCode::value)
-        .map(value -> Byte.toUnsignedInt((byte) value))
-        .max()
-        .orElse(0);
-
-    var values = new PublishAckReasonCode[maxValue + 1];
-
-    for (var value : values()) {
-      values[Byte.toUnsignedInt(value.value)] = value;
-    }
-
-    VALUES = values;
+  public static PublishAckReasonCode ofCode(int code) {
+    return NUMBERED_MAP.require(code);
   }
 
-  public static PublishAckReasonCode ofValue(int value) {
-    return ObjectUtils.notNull(
-        VALUES[value],
-        value,
-        arg -> new IndexOutOfBoundsException("Doesn't support reason code: " + arg));
-  }
+  private final int code;
 
-  byte value;
+  @Override
+  public int number() {
+    return code;
+  }
 }