DefaultPayload copies ByteBuf content

Closes gh-970

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

rsocket-core/src/main/java/io/rsocket/util/DefaultPayload.java

@@ -99,18 +99,18 @@ public static Payload create(ByteBuf data) {
   }
 
   public static Payload create(ByteBuf data, @Nullable ByteBuf metadata) {
-    try {
-      return create(data.nioBuffer(), metadata == null ? null : metadata.nioBuffer());
-    } finally {
-      data.release();
-      if (metadata != null) {
-        metadata.release();
-      }
+    byte[] dataBytes = new byte[data.readableBytes()];
+    data.readBytes(dataBytes);
+    byte[] metadataBytes = null;
+    if (metadata != null) {
+      metadataBytes = new byte[metadata.readableBytes()];
+      metadata.readBytes(metadataBytes);
     }
+    return create(dataBytes, metadataBytes);
   }
 
   public static Payload create(Payload payload) {
-    return create(payload.getData(), payload.hasMetadata() ? payload.getMetadata() : null);
+    return create(payload.data(), payload.hasMetadata() ? payload.metadata() : null);
   }
 
   @Override

rsocket-core/src/test/java/io/rsocket/core/RSocketConnectorTest.java

@@ -73,9 +73,14 @@ public void ensuresThatSetupPayloadCanBeRetained() {
   @Test
   public void ensuresThatMonoFromRSocketConnectorCanBeUsedForMultipleSubscriptions() {
     Payload setupPayload = ByteBufPayload.create("TestData", "TestMetadata");
-
     Assertions.assertThat(setupPayload.refCnt()).isOne();
 
+    // Keep the data and metadata around so we can try changing them independently
+    ByteBuf dataBuf = setupPayload.data();
+    ByteBuf metadataBuf = setupPayload.metadata();
+    dataBuf.retain();
+    metadataBuf.retain();
+
     TestClientTransport testClientTransport = new TestClientTransport();
     Mono<RSocket> connectionMono =
         RSocketConnector.create().setupPayload(setupPayload).connect(testClientTransport);
@@ -92,6 +97,15 @@ public void ensuresThatMonoFromRSocketConnectorCanBeUsedForMultipleSubscriptions
         .expectComplete()
         .verify(Duration.ofMillis(100));
 
+    // Changing the original data and metadata should not impact the SetupPayload
+    dataBuf.writerIndex(dataBuf.readerIndex());
+    dataBuf.writeChar('d');
+    dataBuf.release();
+
+    metadataBuf.writerIndex(metadataBuf.readerIndex());
+    metadataBuf.writeChar('m');
+    metadataBuf.release();
+
     Assertions.assertThat(testClientTransport.testConnection().getSent())
         .hasSize(2)
         .allMatch(
@@ -100,7 +114,11 @@ public void ensuresThatMonoFromRSocketConnectorCanBeUsedForMultipleSubscriptions
               return payload.getDataUtf8().equals("TestData")
                   && payload.getMetadataUtf8().equals("TestMetadata");
             })
-        .allMatch(ReferenceCounted::release);
+        .allMatch(
+            byteBuf -> {
+              System.out.println("calling release " + byteBuf.refCnt());
+              return byteBuf.release();
+            });
     Assertions.assertThat(setupPayload.refCnt()).isZero();
   }
 

rsocket-core/src/test/java/io/rsocket/util/DefaultPayloadTest.java

@@ -103,6 +103,12 @@ public void shouldReleaseGivenByteBufDataAndMetadataUpOnPayloadCreation() {
               metadataPresent
                   ? ByteBuffer.wrap(new byte[] {(byte) (i + 1)})
                   : DefaultPayload.EMPTY_BUFFER);
+
+      data.release();
+      if (metadata != null) {
+        metadata.release();
+      }
+
       allocator.assertHasNoLeaks();
     }
   }