Repository: moquette-io/moquette (java)

Issue:
Shrared buffer issues due to readerIndex
ByteBufs are shared between many sessions and threads. Because if this it is important that none of these session change the `readerIndex` of the shared buffer, and when changing the readerIndex can not be avoided, a `duplicate()` must be made.

In most cases this can be fixed by using the `.getX()` methods instead of the `.readX()` versions.

A PR is on its way.

Pull request description:
Fixed #831: changed several cases where a shared readerIndex was changed
DebugUtils.payload2Str set the readerIndex to the readableBytes, witch was simply wrong. 

Utils.readBytesAndRewind was unused and thus removed. Though it changed the readerIndex, it also reset it, but this use would still have not been thread-safe.

SegmentedPersistentQueueSerDes use of readBytes (that changes the readerIndex) was changed to getBytes which does not change the readerIndex.

Closes #831


Patch:
diff --git a/broker/src/main/java/io/moquette/broker/DebugUtils.java b/broker/src/main/java/io/moquette/broker/DebugUtils.java
index 0fd667ef..9f011ad6 100644
--- a/broker/src/main/java/io/moquette/broker/DebugUtils.java
+++ b/broker/src/main/java/io/moquette/broker/DebugUtils.java
@@ -23,10 +23,7 @@ import java.nio.charset.StandardCharsets;
 public final class DebugUtils {
 
     public static String payload2Str(ByteBuf content) {
-        final int readerPin = content.readableBytes();
-        final String result = content.toString(StandardCharsets.UTF_8);
-        content.readerIndex(readerPin);
-        return result;
+        return content.toString(StandardCharsets.UTF_8);
     }
 
     private DebugUtils() {
diff --git a/broker/src/main/java/io/moquette/broker/Utils.java b/broker/src/main/java/io/moquette/broker/Utils.java
index 1f2ae95a..0107932c 100644
--- a/broker/src/main/java/io/moquette/broker/Utils.java
+++ b/broker/src/main/java/io/moquette/broker/Utils.java
@@ -13,7 +13,6 @@
  *
  * You may elect to redistribute this code under either of these licenses.
  */
-
 package io.moquette.broker;
 
 import io.netty.buffer.ByteBuf;
@@ -42,14 +41,6 @@ public final class Utils {
         return ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
     }
 
-    public static byte[] readBytesAndRewind(ByteBuf payload) {
-        byte[] payloadContent = new byte[payload.readableBytes()];
-        int mark = payload.readerIndex();
-        payload.readBytes(payloadContent);
-        payload.readerIndex(mark);
-        return payloadContent;
-    }
-
     public static MqttVersion versionFromConnect(MqttConnectMessage msg) {
         return MqttVersion.fromProtocolNameAndLevel(msg.variableHeader().name(), (byte) msg.variableHeader().version());
     }
diff --git a/broker/src/main/java/io/moquette/persistence/SegmentedPersistentQueueSerDes.java b/broker/src/main/java/io/moquette/persistence/SegmentedPersistentQueueSerDes.java
index f40df501..6d1f97f1 100644
--- a/broker/src/main/java/io/moquette/persistence/SegmentedPersistentQueueSerDes.java
+++ b/broker/src/main/java/io/moquette/persistence/SegmentedPersistentQueueSerDes.java
@@ -57,7 +57,7 @@ class SegmentedPersistentQueueSerDes {
     private void writePayload(ByteBuffer target, ByteBuf source) {
         final int payloadSize = source.readableBytes();
         byte[] rawBytes = new byte[payloadSize];
-        source.readBytes(rawBytes).release();
+        source.getBytes(source.readerIndex(), rawBytes).release();
         target.putInt(payloadSize);
         target.put(rawBytes);
     }


Test patch:
diff --git a/broker/src/test/java/io/moquette/persistence/SegmentedPersistentQueueSerDesTest.java b/broker/src/test/java/io/moquette/persistence/SegmentedPersistentQueueSerDesTest.java
index ef6b8029..754d8157 100644
--- a/broker/src/test/java/io/moquette/persistence/SegmentedPersistentQueueSerDesTest.java
+++ b/broker/src/test/java/io/moquette/persistence/SegmentedPersistentQueueSerDesTest.java
@@ -19,21 +19,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class SegmentedPersistentQueueSerDesTest {
 
+    private static final String TEST_STRING = "Some fancy things";
+
     @Test
     public void givenEnqueuedMessageWithoutMqttPropertyThenItCanBeProperlySerialized() {
         SegmentedPersistentQueueSerDes sut = new SegmentedPersistentQueueSerDes();
 
         final Topic topic = Topic.asTopic("/metering/temperature");
-        ByteBuf payload = Unpooled.wrappedBuffer("Some fancy things".getBytes(StandardCharsets.UTF_8));
+        ByteBuf payload = Unpooled.wrappedBuffer(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+        payload.retain();
         SessionRegistry.EnqueuedMessage messageToSerialize = new SessionRegistry.PublishedMessage(
             topic, MqttQoS.AT_MOST_ONCE, payload, true, Instant.MAX);
 
         ByteBuffer serialized = sut.toBytes(messageToSerialize);
+        assertEquals(TEST_STRING, payload.toString(StandardCharsets.UTF_8), "Buffer should not have changed.");
+        payload.release();
 
         final SessionRegistry.EnqueuedMessage decoded = sut.fromBytes(serialized);
         assertTrue(decoded instanceof SessionRegistry.PublishedMessage);
         final SessionRegistry.PublishedMessage casted = (SessionRegistry.PublishedMessage) decoded;
         assertEquals(topic, casted.getTopic());
+        assertEquals(TEST_STRING, casted.getPayload().toString(StandardCharsets.UTF_8), "Decoded message not the same.");
     }
 
     @Test
@@ -41,7 +47,7 @@ class SegmentedPersistentQueueSerDesTest {
         SegmentedPersistentQueueSerDes sut = new SegmentedPersistentQueueSerDes();
 
         final Topic topic = Topic.asTopic("/metering/temperature");
-        ByteBuf payload = Unpooled.wrappedBuffer("Some fancy things".getBytes(StandardCharsets.UTF_8));
+        ByteBuf payload = Unpooled.wrappedBuffer(TEST_STRING.getBytes(StandardCharsets.UTF_8));
         int subscriptionId = 123;
         MqttProperties.IntegerProperty intProperty = new MqttProperties.IntegerProperty(
             MqttProperties.MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value(), subscriptionId);