bugfixes resumability and ads related integration tests

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

Author: OlegDokuka

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

@@ -73,11 +73,11 @@ public ClientServerInputMultiplexer(
     this.clientConnection = registry.initConnection(Type.CLIENT, clientReceiver);
   }
 
-  DuplexConnection asResponderConnection() {
+  DuplexConnection asServerConnection() {
     return serverConnection;
   }
 
-  DuplexConnection asRequesterConnection() {
+  DuplexConnection asClientConnection() {
     return clientConnection;
   }
 

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

@@ -3,6 +3,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.rsocket.DuplexConnection;
+import java.nio.channels.ClosedChannelException;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
@@ -24,7 +25,8 @@ class ResumableClientSetup extends ClientSetup {
 
   @Override
   Mono<Tuple2<ByteBuf, DuplexConnection>> init(DuplexConnection connection) {
-    return Mono.create(
-        sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink)));
+    return Mono.<Tuple2<ByteBuf, DuplexConnection>>create(
+            sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink)))
+        .or(connection.onClose().then(Mono.error(ClosedChannelException::new)));
   }
 }

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

@@ -5,6 +5,7 @@
 import io.rsocket.DuplexConnection;
 import io.rsocket.RSocketErrorException;
 import io.rsocket.frame.FrameUtil;
+import java.net.SocketAddress;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
@@ -56,6 +57,11 @@ public ByteBufAllocator alloc() {
     return source.alloc();
   }
 
+  @Override
+  public SocketAddress remoteAddress() {
+    return source.remoteAddress();
+  }
+
   static DuplexConnection wrapIfEnabled(DuplexConnection source) {
     if (LOGGER.isDebugEnabled()) {
       return new LoggingDuplexConnection(source);

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

@@ -621,7 +621,7 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
 
                                   RSocket rSocketRequester =
                                       new RSocketRequester(
-                                          multiplexer.asRequesterConnection(),
+                                          multiplexer.asClientConnection(),
                                           payloadDecoder,
                                           StreamIdSupplier.clientSupplier(),
                                           mtu,
@@ -662,7 +662,7 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
 
                                             RSocket rSocketResponder =
                                                 new RSocketResponder(
-                                                    multiplexer.asResponderConnection(),
+                                                    multiplexer.asServerConnection(),
                                                     wrappedRSocketHandler,
                                                     payloadDecoder,
                                                     responderLeaseHandler,

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

@@ -369,8 +369,7 @@ private Mono<Void> accept(
       default:
         serverSetup.sendError(
             clientServerConnection,
-            new InvalidSetupException(
-                "invalid setup frame: " + FrameHeaderCodec.frameType(startFrame)));
+            new InvalidSetupException("SETUP or RESUME frame must be received before any others"));
         return clientServerConnection.onClose();
     }
   }
@@ -412,7 +411,7 @@ private Mono<Void> acceptSetup(
 
           RSocket rSocketRequester =
               new RSocketRequester(
-                  multiplexer.asResponderConnection(),
+                  multiplexer.asServerConnection(),
                   payloadDecoder,
                   StreamIdSupplier.serverSupplier(),
                   mtu,
@@ -433,7 +432,7 @@ private Mono<Void> acceptSetup(
               .doOnNext(
                   rSocketHandler -> {
                     RSocket wrappedRSocketHandler = interceptors.initResponder(rSocketHandler);
-                    DuplexConnection clientConnection = multiplexer.asRequesterConnection();
+                    DuplexConnection clientConnection = multiplexer.asClientConnection();
 
                     ResponderLeaseHandler responderLeaseHandler =
                         leaseEnabled

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

@@ -27,6 +27,7 @@
 import io.rsocket.frame.SetupFrameCodec;
 import io.rsocket.keepalive.KeepAliveHandler;
 import io.rsocket.resume.*;
+import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -36,8 +37,9 @@
 abstract class ServerSetup {
 
   Mono<Tuple2<ByteBuf, DuplexConnection>> init(DuplexConnection connection) {
-    return Mono.create(
-        sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink)));
+    return Mono.<Tuple2<ByteBuf, DuplexConnection>>create(
+            sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink)))
+        .or(connection.onClose().then(Mono.error(ClosedChannelException::new)));
   }
 
   abstract Mono<Void> acceptRSocketSetup(
@@ -110,14 +112,14 @@ public Mono<Void> acceptRSocketSetup(
             new ResumableDuplexConnection(duplexConnection, resumableFramesStore);
         final ServerRSocketSession serverRSocketSession =
             new ServerRSocketSession(
+                resumeToken,
                 duplexConnection,
                 resumableDuplexConnection,
                 resumeSessionDuration,
                 resumableFramesStore,
-                resumeToken,
                 cleanupStoreOnKeepAlive);
 
-        sessionManager.save(serverRSocketSession);
+        sessionManager.save(serverRSocketSession, resumeToken);
 
         return then.apply(
             new ResumableKeepAliveHandler(

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

@@ -4,6 +4,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.rsocket.DuplexConnection;
 import io.rsocket.RSocketErrorException;
+import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
@@ -42,6 +43,11 @@ public void dispose() {
     source.dispose();
   }
 
+  @Override
+  public boolean isDisposed() {
+    return source.isDisposed();
+  }
+
   @Override
   public Mono<Void> onClose() {
     return source.onClose();
@@ -57,6 +63,11 @@ public Flux<ByteBuf> receive() {
     return this;
   }
 
+  @Override
+  public SocketAddress remoteAddress() {
+    return source.remoteAddress();
+  }
+
   @Override
   public void subscribe(CoreSubscriber<? super ByteBuf> actual) {
     if (done) {