adds docs and descriptions onto resumability related internals

Co-authored-by: Rossen Stoyanchev <[email protected]>
Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java

@@ -592,7 +592,7 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                                         new ClientRSocketSession(
                                             resumeToken,
                                             resumableDuplexConnection,
-                                            connectionMono, // supplies pure
+                                            connectionMono,
                                             resumableClientSetup::init,
                                             resumableFramesStore,
                                             resume.getSessionDuration(),

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

@@ -105,7 +105,7 @@ void reconnect(int index) {
     if (this.s == Operators.cancelledSubscription()
         && S.compareAndSet(this, Operators.cancelledSubscription(), null)) {
       keepAliveSupport.stop();
-      logger.debug("Connection[" + index + "] is lost. Reconnecting...");
+      logger.debug("Connection[" + index + "] is lost. Reconnecting to resume...");
       connectionFactory.retryWhen(retry).timeout(resumeSessionDuration).subscribe(this);
     }
   }

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

@@ -61,8 +61,23 @@ public class InMemoryResumableFramesStore extends Flux<ByteBuf>
 
   CoreSubscriber<? super ByteBuf> actual;
 
-  // indicates whether there is active connection or not
+  /**
+   * Indicates whether there is an active connection or not.
+   *
+   * <ul>
+   *   <li>0 - no active connection
+   *   <li>1 - active connection
+   *   <li>2 - disposed
+   * </ul>
+   *
+   * <pre>
+   * 0 <-----> 1
+   * |         |
+   * +--> 2 <--+
+   * </pre>
+   */
   volatile int state;
+
   static final AtomicIntegerFieldUpdater<InMemoryResumableFramesStore> STATE =
       AtomicIntegerFieldUpdater.newUpdater(InMemoryResumableFramesStore.class, "state");
 
@@ -217,8 +232,6 @@ public void onComplete() {
 
   @Override
   public void onNext(ByteBuf frame) {
-    frame.touch("Tag : " + tag + ". InMemoryResumableFramesStore:onNext");
-
     final int state;
     final boolean isResumable = isResumableFrame(frame);
     if (isResumable) {
@@ -289,7 +302,6 @@ public void subscribe(CoreSubscriber<? super ByteBuf> actual) {
       actual.onSubscribe(this);
       synchronized (this) {
         for (final ByteBuf frame : cachedFrames) {
-          frame.touch("Tag : " + tag + ". InMemoryResumableFramesStore:subscribe");
           actual.onNext(frame.retain());
         }
       }

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

@@ -135,6 +135,11 @@ public void sendFrame(int streamId, ByteBuf frame) {
     }
   }
 
+  /**
+   * Publisher for a sequence of integers starting at 1, with each next number emitted when the
+   * currently active connection is closed and should be resumed. The Publisher never emits an error
+   * and completes when the connection is disposed and not resumed.
+   */
   Flux<Integer> onActiveConnectionClosed() {
     return onConnectionClosedSink.asFlux();
   }
@@ -306,7 +311,6 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(ByteBuf frame) {
-      frame.touch("Tag : " + tag + ". FrameReceivingSubscriber#onNext");
       if (cancelled || s == Operators.cancelledSubscription()) {
         return;
       }

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

@@ -48,6 +48,11 @@ public class ServerRSocketSession
   final ByteBufAllocator allocator;
   final boolean cleanupStoreOnKeepAlive;
 
+  /**
+   * All incoming connections with the Resume intent are enqueued in this queue. Such an approach
+   * ensure that the new connection will affect the resumption state anyhow until the previous
+   * (active) connection is finally closed
+   */
   final Queue<Runnable> connectionsQueue;
 
   volatile int wip;