remove ResumedFramesCalculator

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

Author: mostroverkhov

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

@@ -45,11 +45,7 @@ public ClientRSocketSession(
     this.allocator = Objects.requireNonNull(allocator);
     this.resumableConnection =
         new ResumableDuplexConnection(
-            "client",
-            duplexConnection,
-            ResumedFramesCalculator.ofClient,
-            config.resumeStore(),
-            config.resumeStreamTimeout());
+            "client", duplexConnection, config.resumeStore(), config.resumeStreamTimeout());
 
     resumableConnection
         .connectionErrors()

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

@@ -35,7 +35,6 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
   private static final Logger logger = LoggerFactory.getLogger(ResumableDuplexConnection.class);
 
   private final String tag;
-  private final ResumedFramesCalculator resumedFramesCalculator;
   private final ResumableFramesStore resumableFramesStore;
   private final Duration resumeStreamTimeout;
 
@@ -66,11 +65,9 @@ class ResumableDuplexConnection implements DuplexConnection, ResumeStateHolder {
   ResumableDuplexConnection(
       String tag,
       ResumeAwareConnection duplexConnection,
-      ResumedFramesCalculator resumedFramesCalculator,
       ResumableFramesStore resumableFramesStore,
       Duration resumeStreamTimeout) {
     this.tag = tag;
-    this.resumedFramesCalculator = resumedFramesCalculator;
     this.resumableFramesStore = resumableFramesStore;
     this.resumeStreamTimeout = resumeStreamTimeout;
 
@@ -242,27 +239,34 @@ private void doResume(
     long localPosition = position();
     long localImpliedPosition = impliedPosition();
 
-    logger.debug(
-        "Resumption start. Calculating implied pos using: {}",
-        resumedFramesCalculator.getClass().getSimpleName());
+    logger.debug("Resumption start");
     logger.debug(
         "Resumption states. local: [pos: {}, impliedPos: {}], remote: [pos: {}, impliedPos: {}]",
         localPosition,
         localImpliedPosition,
         remotePosition,
         remoteImpliedPosition);
 
-    Mono<Long> res =
-        resumedFramesCalculator.calculate(
+    long remoteImpliedPos =
+        calculateRemoteImpliedPos(
             localPosition, localImpliedPosition,
             remotePosition, remoteImpliedPosition);
-    Mono<Long> localImpliedPos =
-        res.doOnSuccess(notUsed -> state = State.RESUME)
-            .doOnSuccess(this::releaseFramesToPosition)
-            .map(remoteImpliedPos -> localImpliedPosition);
+
+    Mono<Long> impliedPositionOrError;
+    if (remoteImpliedPos >= 0) {
+      state = State.RESUME;
+      releaseFramesToPosition(remoteImpliedPos);
+      impliedPositionOrError = Mono.just(localImpliedPosition);
+    } else {
+      impliedPositionOrError =
+          Mono.error(
+              new ResumeStateException(
+                  localPosition, localImpliedPosition,
+                  remotePosition, remoteImpliedPos));
+    }
 
     sendResumeFrame
-        .apply(localImpliedPos)
+        .apply(impliedPositionOrError)
         .then(
             streamResumedFrames(
                     resumableFramesStore
@@ -274,6 +278,15 @@ private void doResume(
         .subscribe();
   }
 
+  static long calculateRemoteImpliedPos(
+      long pos, long impliedPos, long remotePos, long remoteImpliedPos) {
+    if (remotePos <= impliedPos && pos <= remoteImpliedPos) {
+      return remoteImpliedPos;
+    } else {
+      return -1L;
+    }
+  }
+
   private void doResumeComplete() {
     logger.debug("Completing resumption");
     state = State.RESUME_COMPLETED;

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

@@ -55,7 +55,6 @@ public ServerRSocketSession(
         new ResumableDuplexConnection(
             "server",
             duplexConnection,
-            ResumedFramesCalculator.ofServer,
             config.resumeStoreFactory().apply(resumeToken),
             config.resumeStreamTimeout());
 

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

@@ -16,56 +16,42 @@
 
 package io.rsocket.resume;
 
-import java.time.Duration;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import reactor.test.StepVerifier;
 
 public class ResumeCalculatorTest {
 
-  private ResumedFramesCalculator clientResumeCalculator;
-  private ResumedFramesCalculator serverResumeCalculator;
-
   @BeforeEach
-  void setUp() {
-    clientResumeCalculator = ResumedFramesCalculator.ofClient;
-    serverResumeCalculator = ResumedFramesCalculator.ofServer;
-  }
+  void setUp() {}
 
   @Test
   void clientResumeSuccess() {
-    StepVerifier.create(clientResumeCalculator.calculate(1, 42, -1, 3))
-        .expectNext(3L)
-        .expectComplete()
-        .verify(Duration.ofSeconds(1));
+    long position = ResumableDuplexConnection.calculateRemoteImpliedPos(1, 42, -1, 3);
+    Assertions.assertEquals(3, position);
   }
 
   @Test
   void clientResumeError() {
-    StepVerifier.create(clientResumeCalculator.calculate(4, 42, -1, 3))
-        .expectError(ResumeStateException.class)
-        .verify(Duration.ofSeconds(1));
+    long position = ResumableDuplexConnection.calculateRemoteImpliedPos(4, 42, -1, 3);
+    Assertions.assertEquals(-1, position);
   }
 
   @Test
   void serverResumeSuccess() {
-    StepVerifier.create(serverResumeCalculator.calculate(1, 42, 4, 23))
-        .expectNext(23L)
-        .expectComplete()
-        .verify(Duration.ofSeconds(1));
+    long position = ResumableDuplexConnection.calculateRemoteImpliedPos(1, 42, 4, 23);
+    Assertions.assertEquals(23, position);
   }
 
   @Test
   void serverResumeErrorClientState() {
-    StepVerifier.create(serverResumeCalculator.calculate(1, 3, 4, 23))
-        .expectError(ResumeStateException.class)
-        .verify(Duration.ofSeconds(1));
+    long position = ResumableDuplexConnection.calculateRemoteImpliedPos(1, 3, 4, 23);
+    Assertions.assertEquals(-1, position);
   }
 
   @Test
   void serverResumeErrorServerState() {
-    StepVerifier.create(serverResumeCalculator.calculate(4, 42, 4, 1))
-        .expectError(ResumeStateException.class)
-        .verify(Duration.ofSeconds(1));
+    long position = ResumableDuplexConnection.calculateRemoteImpliedPos(4, 42, 4, 1);
+    Assertions.assertEquals(-1, position);
   }
 }