fixes ConnectionSetupPayload refcnt management (#854)

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

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

@@ -627,9 +627,10 @@ private Mono<RSocket> connect0(Supplier<ClientTransport> transportSupplier) {
                             mtu);
 
                     return wrappedConnection
-                        .sendOne(setupFrame)
+                        .sendOne(setupFrame.retain())
                         .thenReturn(wrappedRSocketRequester);
-                  });
+                  })
+              .doFinally(signalType -> setup.release());
         });
   }
 }

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

@@ -1,6 +1,9 @@
 package io.rsocket.core;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCounted;
+import io.rsocket.ConnectionSetupPayload;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.test.util.TestClientTransport;
@@ -9,10 +12,60 @@
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
 import reactor.test.StepVerifier;
 
 public class RSocketConnectorTest {
 
+  @Test
+  public void ensuresThatSetupPayloadCanBeRetained() {
+    MonoProcessor<ConnectionSetupPayload> retainedSetupPayload = MonoProcessor.create();
+    TestClientTransport transport = new TestClientTransport();
+
+    ByteBuf data = transport.alloc().buffer();
+
+    data.writeCharSequence("data", CharsetUtil.UTF_8);
+
+    RSocketConnector.create()
+        .setupPayload(ByteBufPayload.create(data))
+        .acceptor(
+            (setup, sendingSocket) -> {
+              retainedSetupPayload.onNext(setup.retain());
+              return Mono.just(new RSocket() {});
+            })
+        .connect(transport)
+        .block();
+
+    Assertions.assertThat(transport.testConnection().getSent())
+        .hasSize(1)
+        .first()
+        .matches(
+            bb -> {
+              DefaultConnectionSetupPayload payload = new DefaultConnectionSetupPayload(bb);
+              return !payload.hasMetadata() && payload.getDataUtf8().equals("data");
+            })
+        .matches(buf -> buf.refCnt() == 2)
+        .matches(
+            buf -> {
+              buf.release();
+              return buf.refCnt() == 1;
+            });
+
+    retainedSetupPayload
+        .as(StepVerifier::create)
+        .expectNextMatches(
+            setup -> {
+              String dataUtf8 = setup.getDataUtf8();
+              return "data".equals(dataUtf8) && setup.release();
+            })
+        .expectComplete()
+        .verify(Duration.ofSeconds(5));
+
+    Assertions.assertThat(retainedSetupPayload.peek().refCnt()).isZero();
+
+    transport.alloc().assertHasNoLeaks();
+  }
+
   @Test
   public void ensuresThatMonoFromRSocketConnectorCanBeUsedForMultipleSubscriptions() {
     Payload setupPayload = ByteBufPayload.create("TestData", "TestMetadata");