bugfix RSocketConnector#setupPayload to copy the content (#843)

Author: OlegDokuka

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

@@ -36,6 +36,7 @@
 import io.rsocket.plugins.InterceptorRegistry;
 import io.rsocket.resume.ClientRSocketSession;
 import io.rsocket.transport.ClientTransport;
+import io.rsocket.util.DefaultPayload;
 import io.rsocket.util.EmptyPayload;
 import java.time.Duration;
 import java.util.Objects;
@@ -121,13 +122,20 @@ public static Mono<RSocket> connectWith(ClientTransport transport) {
    * and metadata should be formatted according to the MIME types specified via {@link
    * #dataMimeType(String)} and {@link #metadataMimeType(String)}.
    *
-   * @param payload the payload containing data and/or metadata for the {@code SETUP} frame
+   * @param payload the payload containing data and/or metadata for the {@code SETUP} frame. Note,
+   *     if the instance of the given payload is not a {@link DefaultPayload}, its content will be
+   *     copied
    * @return the same instance for method chaining
    * @see <a href="https://github.com/rsocket/rsocket/blob/master/Protocol.md#frame-setup">SETUP
    *     Frame</a>
    */
   public RSocketConnector setupPayload(Payload payload) {
-    this.setupPayload = Objects.requireNonNull(payload);
+    if (payload instanceof DefaultPayload) {
+      this.setupPayload = payload;
+    } else {
+      this.setupPayload = DefaultPayload.create(Objects.requireNonNull(payload));
+      payload.release();
+    }
     return this;
   }
 

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

@@ -0,0 +1,49 @@
+package io.rsocket.core;
+
+import io.netty.util.ReferenceCounted;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.test.util.TestClientTransport;
+import io.rsocket.util.ByteBufPayload;
+import java.time.Duration;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+public class RSocketConnectorTest {
+
+  @Test
+  public void ensuresThatMonoFromRSocketConnectorCanBeUsedForMultipleSubscriptions() {
+    Payload setupPayload = ByteBufPayload.create("TestData", "TestMetadata");
+
+    Assertions.assertThat(setupPayload.refCnt()).isOne();
+
+    TestClientTransport testClientTransport = new TestClientTransport();
+    Mono<RSocket> connectionMono =
+        RSocketConnector.create().setupPayload(setupPayload).connect(testClientTransport);
+
+    connectionMono
+        .as(StepVerifier::create)
+        .expectNextCount(1)
+        .expectComplete()
+        .verify(Duration.ofMillis(100));
+
+    connectionMono
+        .as(StepVerifier::create)
+        .expectNextCount(1)
+        .expectComplete()
+        .verify(Duration.ofMillis(100));
+
+    Assertions.assertThat(testClientTransport.testConnection().getSent())
+        .hasSize(2)
+        .allMatch(
+            bb -> {
+              DefaultConnectionSetupPayload payload = new DefaultConnectionSetupPayload(bb);
+              return payload.getDataUtf8().equals("TestData")
+                  && payload.getMetadataUtf8().equals("TestMetadata");
+            })
+        .allMatch(ReferenceCounted::release);
+    Assertions.assertThat(setupPayload.refCnt()).isZero();
+  }
+}