fixed ByteBuffer handling in GenericMqttPublish, copying the bytebuff…

…er on access * fixed MqttPublisherActorTest "acknowledegement" related test errors Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>
eclipse-ditto · Jul 15, 2022 · 6f6f468 · 6f6f468
1 parent 015b2bc
commit 6f6f468
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 5 deletions.
diff --git a/.../ditto/connectivity/service/messaging/mqtt/hivemq/message/publish/GenericMqttPublish.java b/.../ditto/connectivity/service/messaging/mqtt/hivemq/message/publish/GenericMqttPublish.java
@@ -24,6 +24,7 @@
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
 
+import org.eclipse.ditto.base.model.common.ByteBufferUtils;
 import org.eclipse.ditto.base.model.common.CharsetDeterminer;
 import org.eclipse.ditto.base.model.common.ConditionChecker;
 
@@ -328,7 +329,8 @@ public Stream<UserProperty> userProperties() {
 
         @Override
         public Optional<ByteBuffer> getPayload() {
-            return Optional.ofNullable(payload);
+            return Optional.ofNullable(payload)
+                    .map(ByteBufferUtils::clone);
         }
 
         @Override
@@ -446,7 +448,8 @@ public Stream<UserProperty> userProperties() {
 
         @Override
         public Optional<ByteBuffer> getPayload() {
-            return mqtt3Publish.getPayload();
+            return mqtt3Publish.getPayload()
+                    .map(ByteBufferUtils::clone);
         }
 
         @Override
@@ -562,7 +565,8 @@ public Stream<UserProperty> userProperties() {
 
         @Override
         public Optional<ByteBuffer> getPayload() {
-            return mqtt5Publish.getPayload();
+            return mqtt5Publish.getPayload()
+                    .map(ByteBufferUtils::clone);
         }
 
         @Override

diff --git a/...e/ditto/connectivity/service/messaging/mqtt/hivemq/publishing/MqttPublisherActorTest.java b/...e/ditto/connectivity/service/messaging/mqtt/hivemq/publishing/MqttPublisherActorTest.java
@@ -296,7 +296,9 @@ public void sendPublishToBrokerFailureTriggersNegativeAcknowledgementIfAutoAckTa
         final var testKit = new TestKit(actorSystem);
         final var underTest = testKit.childActorOf(getPublisherActorProps());
         final var outboundSignal = OutboundSignalFactory.newMultiMappedOutboundSignal(
-                List.of(getMockOutboundSignalWithAutoAck(AUTO_ACK_LABEL)),
+                List.of(getMockOutboundSignalWithAutoAck(AUTO_ACK_LABEL,
+                        DittoHeaderDefinition.DITTO_ACKREGATOR_ADDRESS.getKey(), testKit.getRef().path().toSerializationFormat()
+                )),
                 testKit.getRef()
         );
         publisherCreated(testKit, underTest);
@@ -348,7 +350,10 @@ public void publishOutboundSignalWithInvalidMqttTopicInHeadersYieldsNegativeAckn
         publisherCreated(testKit, underTest);
 
         final var outboundSignal = OutboundSignalFactory.newMultiMappedOutboundSignal(
-                List.of(getMockOutboundSignalWithAutoAck(AUTO_ACK_LABEL, "custom.topic", invalidMqttTopic)),
+                List.of(getMockOutboundSignalWithAutoAck(AUTO_ACK_LABEL,
+                        "custom.topic", invalidMqttTopic,
+                        DittoHeaderDefinition.DITTO_ACKREGATOR_ADDRESS.getKey(), testKit.getRef().path().toSerializationFormat()
+                )),
                 testKit.getRef()
         );
         underTest.tell(outboundSignal, testKit.getRef());