ResumableDuplexConnection: use drain-queue for actions

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

Author: mostroverkhov

rsocket-core/src/main/java/io/rsocket/keepalive/KeepAliveConnection.java

@@ -33,7 +33,8 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.*;
 
-public class KeepAliveConnection extends DuplexConnectionProxy implements ResumePositionsConnection {
+public class KeepAliveConnection extends DuplexConnectionProxy
+    implements ResumePositionsConnection {
 
   private final MonoProcessor<KeepAliveHandler> keepAliveHandlerReady = MonoProcessor.create();
   private final ByteBufAllocator allocator;

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

@@ -30,7 +30,8 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
-public class ClientRSocketSession implements RSocketSession<Mono<? extends ResumePositionsConnection>> {
+public class ClientRSocketSession
+    implements RSocketSession<Mono<? extends ResumePositionsConnection>> {
   private static final Logger logger = LoggerFactory.getLogger(ClientRSocketSession.class);
 
   private final ResumableDuplexConnection resumableConnection;
@@ -101,7 +102,8 @@ public ClientRSocketSession(
   }
 
   @Override
-  public ClientRSocketSession continueWith(Mono<? extends ResumePositionsConnection> newConnection) {
+  public ClientRSocketSession continueWith(
+      Mono<? extends ResumePositionsConnection> newConnection) {
     this.newConnection = newConnection;
     return this;
   }

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

@@ -22,14 +22,17 @@
 import io.rsocket.frame.FrameHeaderFlyweight;
 import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
+import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.Disposables;
 import reactor.core.publisher.*;
+import reactor.util.concurrent.Queues;
 
 class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
   private static final Logger logger = LoggerFactory.getLogger(ResumableDuplexConnection.class);
@@ -45,7 +48,8 @@ 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 FluxProcessor<Object, Object> actions = UnicastProcessor.create().serialize();
+  private final Queue<Object> actions = Queues.unboundedMultiproducer().get();
+  private final AtomicInteger actionsWip = new AtomicInteger();
 
   private final Mono<Void> framesSent;
   private final RequestListener downStreamRequestListener = new RequestListener();
@@ -56,7 +60,7 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
           128,
           downStreamRequestListener.requests(),
           resumeSaveStreamRequestListener.requests(),
-          actions::onNext);
+          this::dispatch);
 
   private volatile State state;
   private volatile Disposable resumedStreamDisposable = Disposables.disposed();
@@ -92,7 +96,6 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
             .then()
             .cache();
 
-    dispatch(actions);
     reconnect(duplexConnection);
   }
 
@@ -109,7 +112,7 @@ public void reconnect(ResumePositionsConnection connection) {
           "{} Resumable duplex connection reconnected with connection: {}", tag, connection);
       /*race between sendFrame and doResumeStart may lead to ongoing upstream frames
       written before resume complete*/
-      actions.onNext(new ResumeStart(connection));
+      dispatch(new ResumeStart(connection));
     }
   }
 
@@ -118,7 +121,7 @@ public void reconnect(ResumePositionsConnection connection) {
   public void resume(
       long remotePos, long remoteImpliedPos, Function<Mono<Long>, Mono<Void>> resumeFrameSent) {
     /*race between sendFrame and doResume may lead to duplicate frames on resume store*/
-    actions.onNext(new Resume(remotePos, remoteImpliedPos, resumeFrameSent));
+    dispatch(new Resume(remotePos, remoteImpliedPos, resumeFrameSent));
   }
 
   @Override
@@ -214,15 +217,18 @@ 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 dispatch(Object action) {
+    actions.offer(action);
+    if (actionsWip.getAndIncrement() == 0) {
+      do {
+        Object a = actions.poll();
+        if (a instanceof ByteBuf) {
+          sendFrame((ByteBuf) a);
+        } else {
+          ((Runnable) a).run();
+        }
+      } while (actionsWip.decrementAndGet() != 0);
+    }
   }
 
   private void doResumeStart(ResumePositionsConnection connection) {
@@ -272,7 +278,7 @@ private void doResume(
                     resumableFramesStore
                         .resumeStream()
                         .timeout(resumeStreamTimeout)
-                        .doFinally(s -> actions.onNext(new ResumeComplete())))
+                        .doFinally(s -> dispatch(new ResumeComplete())))
                 .doOnError(err -> dispose()))
         .onErrorResume(err -> Mono.empty())
         .subscribe();
@@ -299,7 +305,7 @@ private Mono<Void> streamResumedFrames(Flux<ByteBuf> frames) {
         s -> {
           ResumeFramesSubscriber subscriber =
               new ResumeFramesSubscriber(
-                  downStreamRequestListener.requests(), actions::onNext, s::error, s::success);
+                  downStreamRequestListener.requests(), this::dispatch, s::error, s::success);
           s.onDispose(subscriber);
           resumedStreamDisposable = subscriber;
           frames.subscribe(subscriber);

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

@@ -26,15 +26,12 @@
 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 FluxProcessor<Object, Object> actions = UnicastProcessor.create().serialize();
   private final AtomicBoolean disposed = new AtomicBoolean();
   private final Consumer<ByteBuf> itemConsumer;
   private final Disposable downstreamRequestDisposable;
@@ -59,8 +56,6 @@ class UpstreamFramesSubscriber implements Subscriber<ByteBuf>, Disposable {
 
     resumeSaveStreamDisposable =
         resumeSaveStreamRequests.subscribe(requestN -> requestN(requestN, 0));
-
-    dispatch(actions);
   }
 
   @Override
@@ -75,7 +70,7 @@ public void onSubscribe(Subscription s) {
 
   @Override
   public void onNext(ByteBuf item) {
-    actions.onNext(item);
+    processFrame(item);
   }
 
   @Override
@@ -89,11 +84,17 @@ public void onComplete() {
   }
 
   public void resumeStart() {
-    actions.onNext(new ResumeStart());
+    resumeStarted = true;
   }
 
   public void resumeComplete() {
-    actions.onNext(new ResumeComplete());
+    ByteBuf frame = framesCache.poll();
+    while (frame != null) {
+      itemConsumer.accept(frame);
+      frame = framesCache.poll();
+    }
+    resumeStarted = false;
+    doRequest();
   }
 
   @Override
@@ -113,17 +114,6 @@ 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);
@@ -159,41 +149,11 @@ private void releaseCache() {
     }
   }
 
-  private void doResumeStart() {
-    resumeStarted = true;
-  }
-
-  private void doResumeComplete() {
-    ByteBuf frame = framesCache.poll();
-    while (frame != null) {
-      itemConsumer.accept(frame);
-      frame = framesCache.poll();
-    }
-    resumeStarted = false;
-    doRequest();
-  }
-
   private void processFrame(ByteBuf item) {
     if (resumeStarted) {
       framesCache.offer(item);
     } else {
       itemConsumer.accept(item);
     }
   }
-
-  private class ResumeStart implements Runnable {
-
-    @Override
-    public void run() {
-      doResumeStart();
-    }
-  }
-
-  private class ResumeComplete implements Runnable {
-
-    @Override
-    public void run() {
-      doResumeComplete();
-    }
-  }
 }