refactors DuplexConnection API and reworks resumability impl

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

Author: OlegDokuka

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

@@ -28,13 +28,21 @@ public interface DuplexConnection extends Availability, Closeable {
 
   /**
    * Delivers the given frame to the underlying transport connection. This method is non-blocking
-   * and can be safely executed from multiple threads.
+   * and can be safely executed from multiple threads. This method does not provide any flow-control
+   * mechanism.
    *
    * @param streamId to which the given frame relates
    * @param frame with the encoded content
-   * @param prioritize whether the given frame should be sent with priority to others
    */
-  void sendFrame(int streamId, ByteBuf frame, boolean prioritize);
+  void sendFrame(int streamId, ByteBuf frame);
+
+  /**
+   * Sends terminal frame to the underlying connection and dispose the connection once the frame is
+   * delivered
+   *
+   * @param errorFrame with the encoded error
+   */
+  void sendErrorAndClose(RSocketErrorException errorException);
 
   /**
    * Returns a stream of all {@code Frame}s received on this connection.
@@ -61,12 +69,6 @@ public interface DuplexConnection extends Availability, Closeable {
    */
   Flux<ByteBuf> receive();
 
-  /**
-   * @param errorFrame
-   * @return
-   */
-  void terminate(ByteBuf frame, RSocketErrorException terminalError);
-
   /**
    * Returns the assigned {@link ByteBufAllocator}.
    *

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

@@ -22,14 +22,11 @@
 import io.rsocket.DuplexConnection;
 import io.rsocket.RSocketErrorException;
 import io.rsocket.frame.FrameHeaderCodec;
-import io.rsocket.frame.FrameUtil;
 import io.rsocket.plugins.DuplexConnectionInterceptor.Type;
 import io.rsocket.plugins.InitializingInterceptorRegistry;
 import java.net.SocketAddress;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import org.reactivestreams.Subscription;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -50,67 +47,40 @@
  */
 class ClientServerInputMultiplexer implements CoreSubscriber<ByteBuf>, Closeable {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger("io.rsocket.FrameLogger");
-  private static final InitializingInterceptorRegistry emptyInterceptorRegistry =
-      new InitializingInterceptorRegistry();
-
-  private final InternalDuplexConnection setupReceiver;
   private final InternalDuplexConnection serverReceiver;
   private final InternalDuplexConnection clientReceiver;
-  private final DuplexConnection setupConnection;
   private final DuplexConnection serverConnection;
   private final DuplexConnection clientConnection;
   private final DuplexConnection source;
   private final boolean isClient;
 
   private Subscription s;
-  private boolean setupReceived;
 
   private Throwable t;
 
   private volatile int state;
   private static final AtomicIntegerFieldUpdater<ClientServerInputMultiplexer> STATE =
       AtomicIntegerFieldUpdater.newUpdater(ClientServerInputMultiplexer.class, "state");
 
-  public ClientServerInputMultiplexer(DuplexConnection source) {
-    this(source, emptyInterceptorRegistry, false);
-  }
-
   public ClientServerInputMultiplexer(
       DuplexConnection source, InitializingInterceptorRegistry registry, boolean isClient) {
     this.source = source;
     this.isClient = isClient;
-    source = registry.initConnection(Type.SOURCE, source);
 
-    if (!isClient) {
-      setupReceiver = new InternalDuplexConnection(this, source);
-      setupConnection = registry.initConnection(Type.SETUP, setupReceiver);
-    } else {
-      setupReceiver = null;
-      setupConnection = null;
-    }
-    serverReceiver = new InternalDuplexConnection(this, source);
-    clientReceiver = new InternalDuplexConnection(this, source);
-    serverConnection = registry.initConnection(Type.SERVER, serverReceiver);
-    clientConnection = registry.initConnection(Type.CLIENT, clientReceiver);
+    this.serverReceiver = new InternalDuplexConnection(this, source);
+    this.clientReceiver = new InternalDuplexConnection(this, source);
+    this.serverConnection = registry.initConnection(Type.SERVER, serverReceiver);
+    this.clientConnection = registry.initConnection(Type.CLIENT, clientReceiver);
   }
 
-  DuplexConnection asClientServerConnection() {
-    return source;
-  }
-
-  DuplexConnection asServerConnection() {
+  DuplexConnection asResponderConnection() {
     return serverConnection;
   }
 
-  DuplexConnection asClientConnection() {
+  DuplexConnection asRequesterConnection() {
     return clientConnection;
   }
 
-  DuplexConnection asSetupConnection() {
-    return setupConnection;
-  }
-
   @Override
   public void dispose() {
     source.dispose();
@@ -130,12 +100,7 @@ public Mono<Void> onClose() {
   public void onSubscribe(Subscription s) {
     if (Operators.validate(this.s, s)) {
       this.s = s;
-      if (isClient) {
-        s.request(Long.MAX_VALUE);
-      } else {
-        // request first SetupFrame
-        s.request(1);
-      }
+      s.request(Long.MAX_VALUE);
     }
   }
 
@@ -145,12 +110,6 @@ public void onNext(ByteBuf frame) {
     final Type type;
     if (streamId == 0) {
       switch (FrameHeaderCodec.frameType(frame)) {
-        case SETUP:
-        case RESUME:
-        case RESUME_OK:
-          type = Type.SETUP;
-          setupReceived = true;
-          break;
         case LEASE:
         case KEEPALIVE:
         case ERROR:
@@ -164,19 +123,8 @@ public void onNext(ByteBuf frame) {
     } else {
       type = Type.CLIENT;
     }
-    if (!isClient && type != Type.SETUP && !setupReceived) {
-      final IllegalStateException error =
-          new IllegalStateException("SETUP or LEASE frame must be received before any others.");
-      this.s.cancel();
-      onError(error);
-    }
 
     switch (type) {
-      case SETUP:
-        final InternalDuplexConnection setupReceiver = this.setupReceiver;
-        setupReceiver.onNext(frame);
-        setupReceiver.onComplete();
-        break;
       case CLIENT:
         clientReceiver.onNext(frame);
         break;
@@ -193,16 +141,6 @@ public void onComplete() {
       return;
     }
 
-    if (!isClient) {
-      if (!setupReceived) {
-        setupReceiver.onComplete();
-      }
-
-      if (previousState == 1) {
-        return;
-      }
-    }
-
     if (clientReceiver.isSubscribed()) {
       clientReceiver.onComplete();
     }
@@ -220,16 +158,6 @@ public void onError(Throwable t) {
       return;
     }
 
-    if (!isClient) {
-      if (!setupReceived) {
-        setupReceiver.onError(t);
-      }
-
-      if (previousState == 1) {
-        return;
-      }
-    }
-
     if (clientReceiver.isSubscribed()) {
       clientReceiver.onError(t);
     }
@@ -244,17 +172,8 @@ boolean notifyRequested() {
       return false;
     }
 
-    if (isClient) {
-      if (currentState == 2) {
-        source.receive().subscribe(this);
-      }
-    } else {
-      if (currentState == 1) {
-        source.receive().subscribe(this);
-      } else if (currentState == 3) {
-        // means setup was consumed and we got request from client and server multiplexers
-        s.request(Long.MAX_VALUE);
-      }
+    if (currentState == 2) {
+      source.receive().subscribe(this);
     }
 
     return true;
@@ -280,7 +199,6 @@ private static class InternalDuplexConnection extends Flux<ByteBuf>
       implements Subscription, DuplexConnection {
     private final ClientServerInputMultiplexer clientServerInputMultiplexer;
     private final DuplexConnection source;
-    private final boolean debugEnabled;
 
     private volatile int state;
     static final AtomicIntegerFieldUpdater<InternalDuplexConnection> STATE =
@@ -292,7 +210,6 @@ public InternalDuplexConnection(
         ClientServerInputMultiplexer clientServerInputMultiplexer, DuplexConnection source) {
       this.clientServerInputMultiplexer = clientServerInputMultiplexer;
       this.source = source;
-      this.debugEnabled = LOGGER.isDebugEnabled();
     }
 
     @Override
@@ -340,30 +257,18 @@ void onError(Throwable t) {
     }
 
     @Override
-    public void sendFrame(int streamId, ByteBuf frame, boolean prioritize) {
-      if (debugEnabled) {
-        LOGGER.debug("sending -> " + FrameUtil.toString(frame));
-      }
-
-      source.sendFrame(streamId, frame, prioritize);
+    public void sendFrame(int streamId, ByteBuf frame) {
+      source.sendFrame(streamId, frame);
     }
 
     @Override
-    public void terminate(ByteBuf frame, RSocketErrorException terminalError) {
-      if (debugEnabled) {
-        LOGGER.debug("sending -> " + FrameUtil.toString(frame));
-      }
-
-      source.terminate(frame, terminalError);
+    public void sendErrorAndClose(RSocketErrorException e) {
+      source.sendErrorAndClose(e);
     }
 
     @Override
     public Flux<ByteBuf> receive() {
-      if (debugEnabled) {
-        return this.doOnNext(frame -> LOGGER.debug("receiving -> " + FrameUtil.toString(frame)));
-      } else {
-        return this;
-      }
+      return this;
     }
 
     @Override

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

@@ -0,0 +1,30 @@
+package io.rsocket.core;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.rsocket.DuplexConnection;
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
+
+abstract class ClientSetup {
+  abstract Mono<Tuple2<ByteBuf, DuplexConnection>> init(DuplexConnection connection);
+}
+
+class DefaultClientSetup extends ClientSetup {
+
+  @Override
+  Mono<Tuple2<ByteBuf, DuplexConnection>> init(DuplexConnection connection) {
+    return Mono.create(
+        sink -> sink.onRequest(__ -> sink.success(Tuples.of(Unpooled.EMPTY_BUFFER, connection))));
+  }
+}
+
+class ResumableClientSetup extends ClientSetup {
+
+  @Override
+  Mono<Tuple2<ByteBuf, DuplexConnection>> init(DuplexConnection connection) {
+    return Mono.create(
+        sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink)));
+  }
+}

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

@@ -0,0 +1,66 @@
+package io.rsocket.core;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.rsocket.DuplexConnection;
+import io.rsocket.RSocketErrorException;
+import io.rsocket.frame.FrameUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+class LoggingDuplexConnection implements DuplexConnection {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger("io.rsocket.FrameLogger");
+
+  final DuplexConnection source;
+
+  LoggingDuplexConnection(DuplexConnection source) {
+    this.source = source;
+  }
+
+  @Override
+  public void dispose() {
+    source.dispose();
+  }
+
+  @Override
+  public Mono<Void> onClose() {
+    return source.onClose();
+  }
+
+  @Override
+  public void sendFrame(int streamId, ByteBuf frame) {
+    LOGGER.debug("sending -> " + FrameUtil.toString(frame));
+
+    source.sendFrame(streamId, frame);
+  }
+
+  @Override
+  public void sendErrorAndClose(RSocketErrorException e) {
+    LOGGER.debug("sending -> " + e.getClass().getSimpleName() + ": " + e.getMessage());
+
+    source.sendErrorAndClose(e);
+  }
+
+  @Override
+  public Flux<ByteBuf> receive() {
+    return source
+        .receive()
+        .doOnNext(frame -> LOGGER.debug("receiving -> " + FrameUtil.toString(frame)));
+  }
+
+  @Override
+  public ByteBufAllocator alloc() {
+    return source.alloc();
+  }
+
+  static DuplexConnection wrapIfEnabled(DuplexConnection source) {
+    if (LOGGER.isDebugEnabled()) {
+      return new LoggingDuplexConnection(source);
+    }
+
+    return source;
+  }
+}

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

@@ -109,7 +109,7 @@ public void subscribe(CoreSubscriber<? super Void> actual) {
 
     final ByteBuf requestFrame =
         MetadataPushFrameCodec.encode(this.allocator, metadataRetainedSlice);
-    this.connection.sendFrame(0, requestFrame, true);
+    this.connection.sendFrame(0, requestFrame);
 
     Operators.complete(actual);
   }
@@ -166,7 +166,7 @@ public Void block() {
 
     final ByteBuf requestFrame =
         MetadataPushFrameCodec.encode(this.allocator, metadataRetainedSlice);
-    this.connection.sendFrame(0, requestFrame, true);
+    this.connection.sendFrame(0, requestFrame);
 
     return null;
   }