better performance with resume on

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

Author: mostroverkhov

rsocket-core/src/main/java/io/rsocket/resume/InMemoryResumableFramesStore.java

@@ -29,7 +29,7 @@
 
 public class InMemoryResumableFramesStore implements ResumableFramesStore {
   private static final Logger logger = LoggerFactory.getLogger(InMemoryResumableFramesStore.class);
-  private static final int SAVE_REQUEST_SIZE = 256;
+  private static final long SAVE_REQUEST_SIZE = Long.MAX_VALUE;
 
   private final MonoProcessor<Void> disposed = MonoProcessor.create();
   private volatile long position;
@@ -180,12 +180,12 @@ private static Queue<ByteBuf> cachedFramesQueue(int size) {
   }
 
   private class FramesSubscriber implements Subscriber<ByteBuf> {
-    private final int firstRequestSize;
-    private final int refillSize;
+    private final long firstRequestSize;
+    private final long refillSize;
     private int received;
     private Subscription s;
 
-    public FramesSubscriber(int requestSize) {
+    public FramesSubscriber(long requestSize) {
       this.firstRequestSize = requestSize;
       this.refillSize = firstRequestSize / 2;
     }
@@ -199,7 +199,7 @@ public void onSubscribe(Subscription s) {
     @Override
     public void onNext(ByteBuf byteBuf) {
       saveFrame(byteBuf);
-      if (++received == refillSize) {
+      if (firstRequestSize != Long.MAX_VALUE && ++received == refillSize) {
         received = 0;
         s.request(refillSize);
       }

rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java

@@ -20,7 +20,6 @@
 import io.rsocket.Closeable;
 import io.rsocket.DuplexConnection;
 import io.rsocket.frame.FrameHeaderFlyweight;
-import io.rsocket.internal.UnboundedProcessor;
 import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -47,8 +46,7 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
   private final FluxProcessor<ByteBuf, ByteBuf> downStreamFrames = ReplayProcessor.create(0);
   private final FluxProcessor<ByteBuf, ByteBuf> resumeSaveFrames = EmitterProcessor.create();
   private final MonoProcessor<Void> resumeSaveCompleted = MonoProcessor.create();
-
-  private final UnboundedProcessor<Object> actions = new UnboundedProcessor<>();
+  private final FluxProcessor<Object, Object> actions = UnicastProcessor.create().serialize();
 
   private final Mono<Void> framesSent;
   private final RequestListener downStreamRequestListener = new RequestListener();
@@ -97,12 +95,7 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
             .then()
             .cache();
 
-    Flux<Object> acts = actions.publish().autoConnect(4);
-    acts.ofType(ByteBuf.class).subscribe(this::sendFrame);
-    acts.ofType(ResumeStart.class).subscribe(ResumeStart::run);
-    acts.ofType(Resume.class).subscribe(Resume::run);
-    acts.ofType(ResumeComplete.class).subscribe(ResumeComplete::run);
-
+    dispatch(actions);
     reconnect(duplexConnection);
   }
 
@@ -225,6 +218,17 @@ Flux<Throwable> connectionErrors() {
     return connectionErrors;
   }
 
+  private void dispatch(Flux<?> f) {
+    f.subscribe(
+        o -> {
+          if (o instanceof ByteBuf) {
+            sendFrame((ByteBuf) o);
+          } else {
+            ((Runnable) o).run();
+          }
+        });
+  }
+
   private void doResumeStart(ResumeAwareConnection connection) {
     state = State.RESUME_STARTED;
     resumedStreamDisposable.dispose();

rsocket-core/src/main/java/io/rsocket/resume/UpstreamFramesSubscriber.java

@@ -17,7 +17,6 @@
 package io.rsocket.resume;
 
 import io.netty.buffer.ByteBuf;
-import io.rsocket.internal.UnboundedProcessor;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
@@ -27,13 +26,15 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxProcessor;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.UnicastProcessor;
 import reactor.util.concurrent.Queues;
 
 class UpstreamFramesSubscriber implements Subscriber<ByteBuf>, Disposable {
   private static final Logger logger = LoggerFactory.getLogger(UpstreamFramesSubscriber.class);
 
-  private final UnboundedProcessor<Object> actions = new UnboundedProcessor<>();
+  private final FluxProcessor<Object, Object> actions = UnicastProcessor.create().serialize();
   private final AtomicBoolean disposed = new AtomicBoolean();
   private final Consumer<ByteBuf> itemConsumer;
   private final Disposable downstreamRequestDisposable;
@@ -59,10 +60,7 @@ class UpstreamFramesSubscriber implements Subscriber<ByteBuf>, Disposable {
     resumeSaveStreamDisposable =
         resumeSaveStreamRequests.subscribe(requestN -> requestN(requestN, 0));
 
-    Flux<Object> acts = actions.publish().autoConnect(3);
-    acts.ofType(ByteBuf.class).subscribe(this::processFrame);
-    acts.ofType(ResumeStart.class).subscribe(ResumeStart::run);
-    acts.ofType(ResumeComplete.class).subscribe(ResumeComplete::run);
+    dispatch(actions);
   }
 
   @Override
@@ -115,6 +113,17 @@ public boolean isDisposed() {
     return disposed.get();
   }
 
+  private void dispatch(Flux<?> p) {
+    p.subscribe(
+        o -> {
+          if (o instanceof ByteBuf) {
+            processFrame(((ByteBuf) o));
+          } else {
+            ((Runnable) o).run();
+          }
+        });
+  }
+
   private void requestN(long resumeStreamRequest, long downStreamRequest) {
     synchronized (this) {
       downStreamRequestN = Operators.addCap(downStreamRequestN, downStreamRequest);