enables local transport resumability tests

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

Author: OlegDokuka

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -568,7 +568,7 @@ public TransportPair(
       final RSocketConnector rSocketConnector =
           RSocketConnector.create()
               .payloadDecoder(PayloadDecoder.ZERO_COPY)
-              .keepAlive(Duration.ofMillis(Integer.MAX_VALUE), Duration.ofMillis(Integer.MAX_VALUE))
+              .keepAlive(Duration.ofMillis(10), Duration.ofHours(1))
               .interceptors(
                   registry -> {
                     if (runClientWithAsyncInterceptors && !withResumability) {

rsocket-transport-local/src/main/java/io/rsocket/transport/local/LocalDuplexConnection.java

@@ -136,8 +136,11 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(ByteBuf buf) {
-      actual.onNext(buf);
-      buf.release();
+      try {
+        actual.onNext(buf);
+      } finally {
+        buf.release();
+      }
     }
 
     @Override

rsocket-transport-local/src/test/java/io/rsocket/transport/local/LocalResumableTransportTest.java

@@ -19,9 +19,7 @@
 import io.rsocket.test.TransportTest;
 import java.time.Duration;
 import java.util.UUID;
-import org.junit.jupiter.api.Disabled;
 
-@Disabled("leaking somewhere for no clear reason")
 final class LocalResumableTransportTest implements TransportTest {
 
   private final TransportPair transportPair =