update tcp ping-pong test with resumable transport case

Signed-off-by: Maksym Ostroverkhov <[email protected]>

Author: mostroverkhov

rsocket-examples/src/test/java/io/rsocket/resume/ResumeIntegrationTest.java

@@ -226,7 +226,7 @@ private static Mono<CloseableChannel> newServerRSocket() {
   private static Mono<CloseableChannel> newServerRSocket(int sessionDurationSeconds) {
     return RSocketFactory.receive()
         .resume()
-        .resumeStore(t -> new InMemoryResumableFramesStore("server",100_000))
+        .resumeStore(t -> new InMemoryResumableFramesStore("server", 100_000))
         .resumeSessionDuration(Duration.ofSeconds(sessionDurationSeconds))
         .acceptor((setupPayload, rSocket) -> Mono.just(new TestResponderRSocket()))
         .transport(serverTransport(SERVER_HOST, SERVER_PORT))
@@ -239,14 +239,17 @@ private static class TestResponderRSocket extends AbstractRSocket {
 
     @Override
     public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
-      return duplicate(Flux.interval(Duration.ofMillis(1))
-          .onBackpressureLatest().publishOn(Schedulers.elastic()), 20)
+      return duplicate(
+              Flux.interval(Duration.ofMillis(1))
+                  .onBackpressureLatest()
+                  .publishOn(Schedulers.elastic()),
+              20)
           .map(v -> DefaultPayload.create(String.valueOf(counter.getAndIncrement())))
           .takeUntilOther(Flux.from(payloads).then());
     }
 
     private <T> Flux<T> duplicate(Flux<T> f, int n) {
-      Flux<T> r =Flux.empty();
+      Flux<T> r = Flux.empty();
       for (int i = 0; i < n; i++) {
         r = r.mergeWith(f);
       }

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

@@ -0,0 +1,17 @@
+package io.rsocket.test;
+
+import java.lang.annotation.*;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+/**
+ * {@code @PerfTest} is used to signal that the annotated test class or method is performance test,
+ * and is disabled unless enabled via setting the {@code TEST_PERF_ENABLED} environment variable to
+ * {@code true}.
+ */
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@EnabledIfEnvironmentVariable(named = "TEST_PERF_ENABLED", matches = "(?i)true")
+@Test
+public @interface PerfTest {}

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

@@ -20,7 +20,9 @@
 import io.rsocket.RSocket;
 import io.rsocket.util.ByteBufPayload;
 import java.time.Duration;
+import java.util.function.BiFunction;
 import org.HdrHistogram.Recorder;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -49,16 +51,26 @@ public Recorder startTracker(Duration interval) {
     return histogram;
   }
 
-  public Flux<Payload> startPingPong(int count, final Recorder histogram) {
+  public Flux<Payload> requestResponsePingPong(int count, final Recorder histogram) {
+    return pingPong(RSocket::requestResponse, count, histogram);
+  }
+
+  public Flux<Payload> requestStreamPingPong(int count, final Recorder histogram) {
+    return pingPong(RSocket::requestStream, count, histogram);
+  }
+
+  Flux<Payload> pingPong(
+      BiFunction<RSocket, ? super Payload, ? extends Publisher<Payload>> interaction,
+      int count,
+      final Recorder histogram) {
     return client
         .flatMapMany(
             rsocket ->
                 Flux.range(1, count)
                     .flatMap(
                         i -> {
                           long start = System.nanoTime();
-                          return rsocket
-                              .requestResponse(payload.retain())
+                          return Flux.from(interaction.apply(rsocket, payload.retain()))
                               .doOnNext(Payload::release)
                               .doFinally(
                                   signalType -> {

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

@@ -23,6 +23,7 @@
 import io.rsocket.SocketAcceptor;
 import io.rsocket.util.ByteBufPayload;
 import java.util.concurrent.ThreadLocalRandom;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 public class PingHandler implements SocketAcceptor {
@@ -48,6 +49,12 @@ public Mono<Payload> requestResponse(Payload payload) {
             payload.release();
             return Mono.just(pong.retain());
           }
+
+          @Override
+          public Flux<Payload> requestStream(Payload payload) {
+            payload.release();
+            return Flux.range(0, 100).map(v -> pong.retain());
+          }
         });
   }
 }

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

@@ -48,7 +48,7 @@ public static void main(String... args) {
     int count = 1_000_000_000;
 
     pingClient
-        .startPingPong(count, recorder)
+        .requestResponsePingPong(count, recorder)
         .doOnTerminate(() -> System.out.println("Sent " + count + " messages."))
         .blockLast();
   }

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

@@ -19,30 +19,79 @@
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
 import io.rsocket.frame.decoder.PayloadDecoder;
+import io.rsocket.test.PerfTest;
 import io.rsocket.test.PingClient;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import java.time.Duration;
 import org.HdrHistogram.Recorder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 
+@PerfTest
 public final class TcpPing {
+  private static final int INTERACTIONS_COUNT = 1_000_000_000;
+  private static final int port = Integer.valueOf(System.getProperty("RSOCKET_TEST_PORT", "7878"));
 
-  public static void main(String... args) {
-    Mono<RSocket> client =
-        RSocketFactory.connect()
-            .frameDecoder(PayloadDecoder.ZERO_COPY)
-            .transport(TcpClientTransport.create(7878))
-            .start();
+  @BeforeEach
+  void setUp() {
+    System.out.println("Starting ping-pong test (TCP transport)");
+    System.out.println("port: " + port);
+  }
 
-    PingClient pingClient = new PingClient(client);
+  @Test
+  void requestResponseTest() {
+    PingClient pingClient = newPingClient();
+    Recorder recorder = pingClient.startTracker(Duration.ofSeconds(1));
 
+    pingClient
+        .requestResponsePingPong(INTERACTIONS_COUNT, recorder)
+        .doOnTerminate(() -> System.out.println("Sent " + INTERACTIONS_COUNT + " messages."))
+        .blockLast();
+  }
+
+  @Test
+  void requestStreamTest() {
+    PingClient pingClient = newPingClient();
     Recorder recorder = pingClient.startTracker(Duration.ofSeconds(1));
 
-    int count = 1_000_000_000;
+    pingClient
+        .requestStreamPingPong(INTERACTIONS_COUNT, recorder)
+        .doOnTerminate(() -> System.out.println("Sent " + INTERACTIONS_COUNT + " messages."))
+        .blockLast();
+  }
+
+  @Test
+  void requestStreamResumableTest() {
+    PingClient pingClient = newResumablePingClient();
+    Recorder recorder = pingClient.startTracker(Duration.ofSeconds(1));
 
     pingClient
-        .startPingPong(count, recorder)
-        .doOnTerminate(() -> System.out.println("Sent " + count + " messages."))
+        .requestStreamPingPong(INTERACTIONS_COUNT, recorder)
+        .doOnTerminate(() -> System.out.println("Sent " + INTERACTIONS_COUNT + " messages."))
         .blockLast();
   }
+
+  private static PingClient newPingClient() {
+    return newPingClient(false);
+  }
+
+  private static PingClient newResumablePingClient() {
+    return newPingClient(true);
+  }
+
+  private static PingClient newPingClient(boolean isResumable) {
+    RSocketFactory.ClientRSocketFactory clientRSocketFactory = RSocketFactory.connect();
+    if (isResumable) {
+      clientRSocketFactory.resume();
+    }
+    Mono<RSocket> rSocket =
+        clientRSocketFactory
+            .frameDecoder(PayloadDecoder.ZERO_COPY)
+            .keepAlive(Duration.ofMinutes(1), Duration.ofMinutes(30), 3)
+            .transport(TcpClientTransport.create(port))
+            .start();
+
+    return new PingClient(rSocket);
+  }
 }

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

@@ -22,12 +22,23 @@
 import io.rsocket.transport.netty.server.TcpServerTransport;
 
 public final class TcpPongServer {
+  private static final boolean isResume =
+      Boolean.valueOf(System.getProperty("RSOCKET_TEST_RESUME", "false"));
+  private static final int port = Integer.valueOf(System.getProperty("RSOCKET_TEST_PORT", "7878"));
 
   public static void main(String... args) {
-    RSocketFactory.receive()
+    System.out.println("Starting TCP ping-pong server");
+    System.out.println("port: " + port);
+    System.out.println("resume enabled: " + isResume);
+
+    RSocketFactory.ServerRSocketFactory serverRSocketFactory = RSocketFactory.receive();
+    if (isResume) {
+      serverRSocketFactory.resume();
+    }
+    serverRSocketFactory
         .frameDecoder(PayloadDecoder.ZERO_COPY)
         .acceptor(new PingHandler())
-        .transport(TcpServerTransport.create(7878))
+        .transport(TcpServerTransport.create(port))
         .start()
         .block()
         .onClose()

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

@@ -37,7 +37,7 @@ public static void main(String... args) {
     int count = 1_000_000_000;
 
     pingClient
-        .startPingPong(count, recorder)
+        .requestResponsePingPong(count, recorder)
         .doOnTerminate(() -> System.out.println("Sent " + count + " messages."))
         .blockLast();
   }