Update RSocketPerf.java

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

Author: OlegDokuka

rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java

@@ -22,7 +22,7 @@
 
 @BenchmarkMode(Mode.Throughput)
 @Fork(
-    value = 1 , jvmArgsAppend = {"-Dio.netty.leakDetection.level=advanced"}
+    value = 1 //, jvmArgsAppend = {"-Dio.netty.leakDetection.level=advanced"}
     )
 @Warmup(iterations = 10)
 @Measurement(iterations = 10, time = 10)

rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java

@@ -22,51 +22,65 @@
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
 import io.rsocket.SocketAcceptor;
-import io.rsocket.transport.netty.client.TcpClientTransport;
-import io.rsocket.transport.netty.server.TcpServerTransport;
-import io.rsocket.util.DefaultPayload;
+import io.rsocket.frame.decoder.PayloadDecoder;
+import io.rsocket.transport.local.LocalClientTransport;
+import io.rsocket.transport.local.LocalServerTransport;
+import io.rsocket.util.ByteBufPayload;
 import java.time.Duration;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 public final class ChannelEchoClient {
+  static final Payload payload1 = ByteBufPayload.create("Hello ");
 
   public static void main(String[] args) {
     RSocketFactory.receive()
+        .frameDecoder(PayloadDecoder.ZERO_COPY)
         .acceptor(new SocketAcceptorImpl())
-        .transport(TcpServerTransport.create("localhost", 7000))
+        .transport(LocalServerTransport.create("localhost"))
         .start()
         .subscribe();
 
     RSocket socket =
         RSocketFactory.connect()
-            .transport(TcpClientTransport.create("localhost", 7000))
+            .keepAliveAckTimeout(Duration.ofMinutes(10))
+            .frameDecoder(PayloadDecoder.ZERO_COPY)
+            .transport(LocalClientTransport.create("localhost"))
             .start()
             .block();
 
-    socket
-        .requestChannel(
-            Flux.interval(Duration.ofMillis(1000)).map(i -> DefaultPayload.create("Hello")))
-        .map(Payload::getDataUtf8)
-        .doOnNext(System.out::println)
-        .take(10)
-        .doFinally(signalType -> socket.dispose())
-        .then()
-        .block();
+    Flux.range(0, 100000000)
+        .concatMap(i -> socket.fireAndForget(payload1.retain()))
+        //        .doOnNext(p -> {
+        ////            System.out.println(p.getDataUtf8());
+        //            p.release();
+        //        })
+        .blockLast();
   }
 
   private static class SocketAcceptorImpl implements SocketAcceptor {
     @Override
     public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {
       return Mono.just(
           new AbstractRSocket() {
+
+            @Override
+            public Mono<Void> fireAndForget(Payload payload) {
+              //                  System.out.println(payload.getDataUtf8());
+              payload.release();
+              return Mono.empty();
+            }
+
+            @Override
+            public Mono<Payload> requestResponse(Payload payload) {
+              return Mono.just(payload);
+            }
+
             @Override
             public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
-              return Flux.from(payloads)
-                  .map(Payload::getDataUtf8)
-                  .map(s -> "Echo: " + s)
-                  .map(DefaultPayload::create);
+              return Flux.from(payloads).subscribeOn(Schedulers.single());
             }
           });
     }