improves transport resumability tests

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

Author: OlegDokuka

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

@@ -547,7 +547,7 @@ public TransportPair(
                                       "Server",
                                       duplexConnection,
                                       Duration.ofMillis(
-                                          ThreadLocalRandom.current().nextInt(10, 1500)))
+                                          ThreadLocalRandom.current().nextInt(100, 500)))
                                   : duplexConnection);
                     }
                   });
@@ -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) {
@@ -594,7 +594,7 @@ public TransportPair(
                                       "Client",
                                       duplexConnection,
                                       Duration.ofMillis(
-                                          ThreadLocalRandom.current().nextInt(1, 2000)))
+                                          ThreadLocalRandom.current().nextInt(100, 500)))
                                   : duplexConnection);
                     }
                   });

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 =
@@ -34,7 +32,7 @@ final class LocalResumableTransportTest implements TransportTest {
 
   @Override
   public Duration getTimeout() {
-    return Duration.ofSeconds(10);
+    return Duration.ofMinutes(3);
   }
 
   @Override

rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/WebsocketResumableTransportTest.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2015-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.transport.netty;
+
+import io.netty.channel.ChannelOption;
+import io.rsocket.test.TransportTest;
+import io.rsocket.transport.netty.client.WebsocketClientTransport;
+import io.rsocket.transport.netty.server.WebsocketServerTransport;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.http.server.HttpServer;
+
+final class WebsocketResumableTransportTest implements TransportTest {
+
+  private final TransportPair transportPair =
+      new TransportPair<>(
+          () -> InetSocketAddress.createUnresolved("localhost", 0),
+          (address, server, allocator) ->
+              WebsocketClientTransport.create(
+                  HttpClient.create()
+                      .host(server.address().getHostName())
+                      .port(server.address().getPort())
+                      .option(ChannelOption.ALLOCATOR, allocator),
+                  ""),
+          (address, allocator) ->
+              WebsocketServerTransport.create(
+                  HttpServer.create()
+                      .host(address.getHostName())
+                      .port(address.getPort())
+                      .option(ChannelOption.ALLOCATOR, allocator)),
+          false,
+          true);
+
+  @Override
+  public Duration getTimeout() {
+    return Duration.ofMinutes(3);
+  }
+
+  @Override
+  public TransportPair getTransportPair() {
+    return transportPair;
+  }
+}