Ensure setupPayload is actually copied

Closes gh-970

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

Author: rstoyanchev

rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -153,12 +153,24 @@ public RSocketConnector setupPayload(Payload payload) {
     if (payload instanceof DefaultPayload) {
       this.setupPayloadMono = Mono.just(payload);
     } else {
-      this.setupPayloadMono = Mono.just(DefaultPayload.create(Objects.requireNonNull(payload)));
+      this.setupPayloadMono = Mono.just(copyPayload(payload));
       payload.release();
     }
     return this;
   }
 
+  private Payload copyPayload(Payload payload) {
+    Objects.requireNonNull(payload);
+    byte[] dataBytes = new byte[payload.data().readableBytes()];
+    payload.data().readBytes(dataBytes);
+    byte[] metadataBytes = null;
+    if (payload.metadata().readableBytes() > 0) {
+      metadataBytes = new byte[payload.metadata().readableBytes()];
+      payload.metadata().readBytes(metadataBytes);
+    }
+    return DefaultPayload.create(dataBytes, metadataBytes);
+  }
+
   /**
    * Set the MIME type to use for formatting payload data on the established connection. This is set
    * in the initial {@code SETUP} frame sent to the server.

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();
   }