work on publish storage

Author: JavaSaBr

core-service/src/main/java/javasabr/mqtt/service/impl/InMemoryRetainPublishService.java

@@ -21,7 +21,7 @@ public InMemoryRetainPublishService() {
   @Override
   public void retain(Publish publish) {
     PublishData data = publish.data();
-    if (data.isEmpty()) {
+    if (data.isPayloadEmpty()) {
       retainedMessageTree.removeRetainedMessage(publish.topicName());
     } else {
       retainedMessageTree.addRetainedMessage(publish);

core-service/src/main/java/javasabr/mqtt/service/message/out/factory/Mqtt311MessageOutFactory.java

@@ -1,9 +1,9 @@
 package javasabr.mqtt.service.message.out.factory;
 
 import javasabr.mqtt.model.MqttVersion;
-import javasabr.mqtt.model.PayloadFormat;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.data.type.StringPair;
+import javasabr.mqtt.model.publish.PublishData;
 import javasabr.mqtt.model.reason.code.AuthenticateReasonCode;
 import javasabr.mqtt.model.reason.code.ConnectAckReasonCode;
 import javasabr.mqtt.model.reason.code.DisconnectReasonCode;
@@ -63,12 +63,10 @@ public MqttOutMessage newPublish(
       boolean duplicate,
       TopicName topicName,
       int topicAlias,
-      byte[] payload,
-      PayloadFormat payloadFormat,
+      PublishData data,
       @Nullable TopicName responseTopic,
-      byte @Nullable [] correlationData,
       Array<StringPair> userProperties) {
-    return new PublishMqtt311OutMessage(messageId, qos, retain, duplicate, topicName, payload);
+    return new PublishMqtt311OutMessage(messageId, qos, retain, duplicate, topicName, data);
   }
 
   @Override

core-service/src/main/java/javasabr/mqtt/service/message/out/factory/Mqtt5MessageOutFactory.java

@@ -2,9 +2,9 @@
 
 import javasabr.mqtt.model.MqttClientConnectionConfig;
 import javasabr.mqtt.model.MqttVersion;
-import javasabr.mqtt.model.PayloadFormat;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.data.type.StringPair;
+import javasabr.mqtt.model.publish.PublishData;
 import javasabr.mqtt.model.reason.code.AuthenticateReasonCode;
 import javasabr.mqtt.model.reason.code.ConnectAckReasonCode;
 import javasabr.mqtt.model.reason.code.DisconnectReasonCode;
@@ -76,22 +76,18 @@ public MqttOutMessage newPublish(
       boolean duplicate,
       TopicName topicName,
       int topicAlias,
-      byte[] payload,
-      PayloadFormat payloadFormat,
+      PublishData data,
       @Nullable TopicName responseTopic,
-      byte @Nullable [] correlationData,
       Array<StringPair> userProperties) {
     return new PublishMqtt5OutMessage(
         messageId,
         qos,
         retain,
         duplicate,
         topicName,
-        payload,
+        data,
         topicAlias,
-        payloadFormat,
         responseTopic,
-        correlationData,
         userProperties);
   }
 

core-service/src/main/java/javasabr/mqtt/service/publish/impl/InMemoryPublishData.java

@@ -40,10 +40,15 @@ public InMemoryPublishData(
   }
 
   @Override
-  public boolean isEmpty() {
+  public boolean isPayloadEmpty() {
     return payloadSize < 1;
   }
 
+  @Override
+  public boolean isCorrelationDataEmpty() {
+    return correlationDataSize < 1;
+  }
+
   @Override
   public void writePayloadTo(ByteBuffer buffer) {
     buffer.put(payload);

core-service/src/main/java/javasabr/mqtt/service/session/impl/InMemoryProcessingPublishes.java

@@ -3,7 +3,7 @@
 import java.util.concurrent.locks.StampedLock;
 import javasabr.mqtt.model.MqttUser;
 import javasabr.mqtt.model.message.TrackableMqttMessage;
-import javasabr.mqtt.model.publishing.Publish;
+import javasabr.mqtt.model.publish.Publish;
 import javasabr.mqtt.model.session.MqttSession;
 import javasabr.mqtt.model.session.ProcessingPublishes;
 import javasabr.mqtt.model.session.PublishRetryer;

model/src/main/java/javasabr/mqtt/model/publish/PublishData.java

@@ -16,7 +16,8 @@ public interface PublishData {
   int payloadSize();
   int correlationDataSize();
 
-  boolean isEmpty();
+  boolean isPayloadEmpty();
+  boolean isCorrelationDataEmpty();
 
   void writePayloadTo(ByteBuffer buffer);
   void writeCorrelationDataTo(ByteBuffer buffer);

network/src/main/java/javasabr/mqtt/network/message/out/PublishMqtt311OutMessage.java

@@ -4,6 +4,7 @@
 import javasabr.mqtt.base.util.DebugUtils;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.message.MqttMessageType;
+import javasabr.mqtt.model.publish.PublishData;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.mqtt.network.MqttConnection;
 import lombok.AccessLevel;
@@ -23,7 +24,7 @@ public class PublishMqtt311OutMessage extends TrackableMqttOutMessage {
   }
 
   QoS qos;
-  byte[] payload;
+  PublishData data;
   TopicName topicName;
 
   boolean retain;
@@ -35,12 +36,12 @@ public PublishMqtt311OutMessage(
       boolean retain,
       boolean duplicate,
       TopicName topicName,
-      byte[] payload) {
+      PublishData data) {
     super(messageId);
     this.qos = qos;
     this.retain = retain;
     this.duplicate = duplicate;
-    this.payload = payload;
+    this.data = data;
     this.topicName = topicName;
   }
 
@@ -56,7 +57,7 @@ public MqttMessageType messageType() {
 
   @Override
   public int expectedLength(MqttConnection connection) {
-    return 7 + payload.length;
+    return 7 + data.payloadSize();
   }
 
   @Override
@@ -83,6 +84,6 @@ protected void writeVariableHeader(MqttConnection connection, ByteBuffer buffer)
   @Override
   protected void writePayload(MqttConnection connection, ByteBuffer buffer) {
     // https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc384800413
-    buffer.put(payload);
+    data.writePayloadTo(buffer);
   }
 }

network/src/main/java/javasabr/mqtt/network/message/out/PublishMqtt5OutMessage.java

@@ -6,9 +6,9 @@
 import javasabr.mqtt.base.util.DebugUtils;
 import javasabr.mqtt.model.MqttMessageProperty;
 import javasabr.mqtt.model.MqttProperties;
-import javasabr.mqtt.model.PayloadFormat;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.data.type.StringPair;
+import javasabr.mqtt.model.publish.PublishData;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.mqtt.network.MqttConnection;
 import javasabr.rlib.collections.array.Array;
@@ -148,10 +148,8 @@ public class PublishMqtt5OutMessage extends PublishMqtt311OutMessage {
 
   @Nullable
   TopicName responseTopic;
-  byte @Nullable [] correlationData;
 
   Array<StringPair> userProperties;
-  PayloadFormat payloadFormat;
 
   int topicAlias;
 
@@ -161,17 +159,13 @@ public PublishMqtt5OutMessage(
       boolean retain,
       boolean duplicate,
       TopicName topicName,
-      byte[] payload,
+      PublishData data,
       int topicAlias,
-      PayloadFormat payloadFormat,
       @Nullable TopicName responseTopic,
-      byte @Nullable [] correlationData,
       Array<StringPair> userProperties) {
-    super(messageId, qos, retain, duplicate, topicName, payload);
+    super(messageId, qos, retain, duplicate, topicName, data);
     this.topicAlias = topicAlias;
-    this.payloadFormat = payloadFormat;
     this.responseTopic = responseTopic;
-    this.correlationData = correlationData;
     this.userProperties = userProperties;
   }
 
@@ -189,15 +183,19 @@ protected boolean isPropertiesSupported(MqttConnection connection) {
   protected void writeProperties(MqttConnection connection, ByteBuffer buffer) {
     // https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc511988586
     String rawResponseTopic = responseTopic == null ? null : responseTopic.rawTopic();
-    writeProperty(buffer, MqttMessageProperty.PAYLOAD_FORMAT_INDICATOR, payloadFormat.code());
+    writeProperty(buffer, MqttMessageProperty.PAYLOAD_FORMAT_INDICATOR, data.payloadFormat().code());
     writeProperty(
         buffer,
         MqttMessageProperty.MESSAGE_EXPIRY_INTERVAL,
         0,
         MqttProperties.MESSAGE_EXPIRY_INTERVAL_IS_NOT_SET);
     writeProperty(buffer, MqttMessageProperty.TOPIC_ALIAS, topicAlias, MqttProperties.TOPIC_ALIAS_NOT_SET);
     writeNotEmptyProperty(buffer, MqttMessageProperty.RESPONSE_TOPIC, rawResponseTopic);
-    writeNotEmptyProperty(buffer, MqttMessageProperty.CORRELATION_DATA, correlationData);
     writeStringPairProperties(buffer, MqttMessageProperty.USER_PROPERTY, userProperties);
+
+    if (!data.isCorrelationDataEmpty()) {
+      buffer.put(MqttMessageProperty.CORRELATION_DATA.id());
+      data.writeCorrelationDataTo(buffer);
+    }
   }
 }