provides enhanced DuplexConnection api

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

Author: OlegDokuka

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

@@ -20,40 +20,21 @@
 import io.netty.buffer.ByteBufAllocator;
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
-import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 /** Represents a connection with input/output that the protocol uses. */
 public interface DuplexConnection extends Availability, Closeable {
 
   /**
-   * Sends the source of Frames on this connection and returns the {@code Publisher} representing
-   * the result of this send.
+   * Delivers the given frame to the underlying transport connection. This method is non-blocking
+   * and can be safely executed from multiple threads.
    *
-   * <p><strong>Flow control</strong>
-   *
-   * <p>The passed {@code Publisher} must
-   *
-   * @param frames Stream of {@code Frame}s to send on the connection.
-   * @return {@code Publisher} that completes when all the frames are written on the connection
-   *     successfully and errors when it fails.
-   * @throws NullPointerException if {@code frames} is {@code null}
-   */
-  Mono<Void> send(Publisher<ByteBuf> frames);
-
-  /**
-   * Sends a single {@code Frame} on this connection and returns the {@code Publisher} representing
-   * the result of this send.
-   *
-   * @param frame {@code Frame} to send.
-   * @return {@code Publisher} that completes when the frame is written on the connection
-   *     successfully and errors when it fails.
+   * @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
    */
-  default Mono<Void> sendOne(ByteBuf frame) {
-    return send(Mono.just(frame));
-  }
+  void sendFrame(int streamId, ByteBuf frame, boolean prioritize);
 
   /**
    * Returns a stream of all {@code Frame}s received on this connection.
@@ -80,6 +61,12 @@ default Mono<Void> sendOne(ByteBuf frame) {
    */
   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

@@ -20,13 +20,13 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.rsocket.Closeable;
 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.Publisher;
 import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -95,19 +95,19 @@ public ClientServerInputMultiplexer(
     clientConnection = registry.initConnection(Type.CLIENT, clientReceiver);
   }
 
-  public DuplexConnection asClientServerConnection() {
+  DuplexConnection asClientServerConnection() {
     return source;
   }
 
-  public DuplexConnection asServerConnection() {
+  DuplexConnection asServerConnection() {
     return serverConnection;
   }
 
-  public DuplexConnection asClientConnection() {
+  DuplexConnection asClientConnection() {
     return clientConnection;
   }
 
-  public DuplexConnection asSetupConnection() {
+  DuplexConnection asSetupConnection() {
     return setupConnection;
   }
 
@@ -340,23 +340,21 @@ void onError(Throwable t) {
     }
 
     @Override
-    public Mono<Void> send(Publisher<ByteBuf> frame) {
+    public void sendFrame(int streamId, ByteBuf frame, boolean prioritize) {
       if (debugEnabled) {
-        return Flux.from(frame)
-            .doOnNext(f -> LOGGER.debug("sending -> " + FrameUtil.toString(f)))
-            .as(source::send);
+        LOGGER.debug("sending -> " + FrameUtil.toString(frame));
       }
 
-      return source.send(frame);
+      source.sendFrame(streamId, frame, prioritize);
     }
 
     @Override
-    public Mono<Void> sendOne(ByteBuf frame) {
+    public void terminate(ByteBuf frame, RSocketErrorException terminalError) {
       if (debugEnabled) {
         LOGGER.debug("sending -> " + FrameUtil.toString(frame));
       }
 
-      return source.sendOne(frame);
+      source.terminate(frame, terminalError);
     }
 
     @Override

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

@@ -20,12 +20,11 @@
 import static io.rsocket.core.SendUtils.sendReleasingPayload;
 import static io.rsocket.core.StateUtils.*;
 
-import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.IllegalReferenceCountException;
+import io.rsocket.DuplexConnection;
 import io.rsocket.Payload;
 import io.rsocket.frame.FrameType;
-import io.rsocket.internal.UnboundedProcessor;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.reactivestreams.Subscription;
@@ -50,15 +49,15 @@ final class FireAndForgetRequesterMono extends Mono<Void> implements Subscriptio
   final int mtu;
   final int maxFrameLength;
   final RequesterResponderSupport requesterResponderSupport;
-  final UnboundedProcessor<ByteBuf> sendProcessor;
+  final DuplexConnection connection;
 
   FireAndForgetRequesterMono(Payload payload, RequesterResponderSupport requesterResponderSupport) {
     this.allocator = requesterResponderSupport.getAllocator();
     this.payload = payload;
     this.mtu = requesterResponderSupport.getMtu();
     this.maxFrameLength = requesterResponderSupport.getMaxFrameLength();
     this.requesterResponderSupport = requesterResponderSupport;
-    this.sendProcessor = requesterResponderSupport.getSendProcessor();
+    this.connection = requesterResponderSupport.getDuplexConnection();
   }
 
   @Override
@@ -106,7 +105,7 @@ public void subscribe(CoreSubscriber<? super Void> actual) {
       }
 
       sendReleasingPayload(
-          streamId, FrameType.REQUEST_FNF, mtu, p, this.sendProcessor, this.allocator, true);
+          streamId, FrameType.REQUEST_FNF, mtu, p, this.connection, this.allocator, true);
     } catch (Throwable e) {
       lazyTerminate(STATE, this);
       actual.onError(e);
@@ -169,7 +168,7 @@ public Void block() {
           FrameType.REQUEST_FNF,
           this.mtu,
           this.payload,
-          this.sendProcessor,
+          this.connection,
           this.allocator,
           true);
     } catch (Throwable e) {

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

@@ -22,9 +22,9 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.IllegalReferenceCountException;
+import io.rsocket.DuplexConnection;
 import io.rsocket.Payload;
 import io.rsocket.frame.MetadataPushFrameCodec;
-import io.rsocket.internal.UnboundedProcessor;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import reactor.core.CoreSubscriber;
@@ -43,13 +43,13 @@ final class MetadataPushRequesterMono extends Mono<Void> implements Scannable {
   final ByteBufAllocator allocator;
   final Payload payload;
   final int maxFrameLength;
-  final UnboundedProcessor<ByteBuf> sendProcessor;
+  final DuplexConnection connection;
 
   MetadataPushRequesterMono(Payload payload, RequesterResponderSupport requesterResponderSupport) {
     this.allocator = requesterResponderSupport.getAllocator();
     this.payload = payload;
     this.maxFrameLength = requesterResponderSupport.getMaxFrameLength();
-    this.sendProcessor = requesterResponderSupport.getSendProcessor();
+    this.connection = requesterResponderSupport.getDuplexConnection();
   }
 
   @Override
@@ -109,7 +109,7 @@ public void subscribe(CoreSubscriber<? super Void> actual) {
 
     final ByteBuf requestFrame =
         MetadataPushFrameCodec.encode(this.allocator, metadataRetainedSlice);
-    this.sendProcessor.onNext(requestFrame);
+    this.connection.sendFrame(0, requestFrame, true);
 
     Operators.complete(actual);
   }
@@ -166,7 +166,7 @@ public Void block() {
 
     final ByteBuf requestFrame =
         MetadataPushFrameCodec.encode(this.allocator, metadataRetainedSlice);
-    this.sendProcessor.onNext(requestFrame);
+    this.connection.sendFrame(0, requestFrame, true);
 
     return null;
   }

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

@@ -551,8 +551,10 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                                   .resumeToken(resumeToken);
                           keepAliveHandler =
                               new KeepAliveHandler.ResumableKeepAliveHandler(
-                                  session.resumableConnection());
-                          wrappedConnection = session.resumableConnection();
+                                  /*session.resumableConnection()*/ null);
+                          //                          wrappedConnection =
+                          // session.resumableConnection();
+                          wrappedConnection = null;
                         } else {
                           resumeToken = Unpooled.EMPTY_BUFFER;
                           keepAliveHandler =
@@ -608,7 +610,7 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                         return interceptors
                             .initSocketAcceptor(acceptor)
                             .accept(setup, wrappedRSocketRequester)
-                            .flatMap(
+                            .map(
                                 rSocketHandler -> {
                                   RSocket wrappedRSocketHandler =
                                       interceptors.initResponder(rSocketHandler);
@@ -632,9 +634,9 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                                           maxFrameLength,
                                           maxInboundPayloadSize);
 
-                                  return wrappedConnection
-                                      .sendOne(setupFrame.retain())
-                                      .thenReturn(wrappedRSocketRequester);
+                                  wrappedConnection.sendFrame(0, setupFrame.retain(), false);
+
+                                  return wrappedRSocketRequester;
                                 })
                             .doFinally(signalType -> setup.release());
                       });

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

@@ -29,7 +29,6 @@
 import io.rsocket.frame.FrameType;
 import io.rsocket.frame.RequestNFrameCodec;
 import io.rsocket.frame.decoder.PayloadDecoder;
-import io.rsocket.internal.UnboundedProcessor;
 import io.rsocket.keepalive.KeepAliveFramesAcceptor;
 import io.rsocket.keepalive.KeepAliveHandler;
 import io.rsocket.keepalive.KeepAliveSupport;
@@ -62,7 +61,6 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
       AtomicReferenceFieldUpdater.newUpdater(
           RSocketRequester.class, Throwable.class, "terminationError");
 
-  private final DuplexConnection connection;
   private final RequesterLeaseHandler leaseHandler;
   private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
   private final MonoProcessor<Void> onClose;
@@ -78,23 +76,13 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
       int keepAliveAckTimeout,
       @Nullable KeepAliveHandler keepAliveHandler,
       RequesterLeaseHandler leaseHandler) {
-    super(
-        mtu,
-        maxFrameLength,
-        maxInboundPayloadSize,
-        payloadDecoder,
-        connection.alloc(),
-        streamIdSupplier);
-
-    this.connection = connection;
+    super(mtu, maxFrameLength, maxInboundPayloadSize, payloadDecoder, connection, streamIdSupplier);
+
     this.leaseHandler = leaseHandler;
     this.onClose = MonoProcessor.create();
 
-    UnboundedProcessor<ByteBuf> sendProcessor = super.getSendProcessor();
-
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     connection.onClose().subscribe(null, this::tryTerminateOnConnectionError, this::tryShutdown);
-    connection.send(sendProcessor).subscribe(null, this::handleSendProcessorError);
 
     connection.receive().subscribe(this::handleIncomingFrames, e -> {});
 
@@ -103,7 +91,9 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
           new ClientKeepAliveSupport(this.getAllocator(), keepAliveTickPeriod, keepAliveAckTimeout);
       this.keepAliveFramesAcceptor =
           keepAliveHandler.start(
-              keepAliveSupport, sendProcessor::onNextPrioritized, this::tryTerminateOnKeepAlive);
+              keepAliveSupport,
+              (keepAliveFrame) -> connection.sendFrame(0, keepAliveFrame, true),
+              this::tryTerminateOnKeepAlive);
     } else {
       keepAliveFramesAcceptor = null;
     }
@@ -177,7 +167,7 @@ public int addAndGetNextStreamId(FrameHandler frameHandler) {
 
   @Override
   public double availability() {
-    return Math.min(connection.availability(), leaseHandler.availability());
+    return Math.min(getDuplexConnection().availability(), leaseHandler.availability());
   }
 
   @Override
@@ -206,13 +196,10 @@ private void handleIncomingFrames(ByteBuf frame) {
       }
     } catch (Throwable t) {
       LOGGER.error("Unexpected error during frame handling", t);
-      super.getSendProcessor()
-          .onNext(
-              ErrorFrameCodec.encode(
-                  super.getAllocator(),
-                  0,
-                  new ConnectionErrorException("Unexpected error during frame handling", t)));
-      this.tryTerminateOnConnectionError(t);
+      final ConnectionErrorException error =
+          new ConnectionErrorException("Unexpected error during frame handling", t);
+      getDuplexConnection()
+          .terminate(ErrorFrameCodec.encode(super.getAllocator(), 0, error), error);
     }
   }
 
@@ -332,7 +319,7 @@ private void terminate(Throwable e) {
     if (keepAliveFramesAcceptor != null) {
       keepAliveFramesAcceptor.dispose();
     }
-    connection.dispose();
+    getDuplexConnection().dispose();
     leaseHandler.dispose();
 
     synchronized (this) {
@@ -347,15 +334,10 @@ private void terminate(Throwable e) {
               });
     }
 
-    this.getSendProcessor().dispose();
     if (e == CLOSED_CHANNEL_EXCEPTION) {
       onClose.onComplete();
     } else {
       onClose.onError(e);
     }
   }
-
-  private void handleSendProcessorError(Throwable t) {
-    connection.dispose();
-  }
 }