FTTF-git
diff --git a/‎rsocket-core/src/main/java/io/rsocket/RSocketFactory.java
Lines changed: 28 additions & 13 deletions b/‎rsocket-core/src/main/java/io/rsocket/RSocketFactory.java
Lines changed: 28 additions & 13 deletions
@@ -42,6 +42,7 @@
 import io.rsocket.util.EmptyPayload;
 import java.time.Duration;
 import java.util.Objects;
+import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -94,6 +95,8 @@ public static class ClientRSocketFactory implements ClientTransportAcceptor {
     private Supplier<Function<RSocket, RSocket>> acceptor =
         () -> rSocket -> new AbstractRSocket() {};
 
+    private BiFunction<ConnectionSetupPayload, RSocket, RSocket> biAcceptor;
+
     private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
     private int mtu = 0;
     private PluginRegistry plugins = new PluginRegistry(Plugins.defaultPlugins());
@@ -242,6 +245,12 @@ public ClientTransportAcceptor acceptor(Supplier<Function<RSocket, RSocket>> acc
       return StartClient::new;
     }
 
+    public ClientTransportAcceptor acceptor(
+        BiFunction<ConnectionSetupPayload, RSocket, RSocket> biAcceptor) {
+      this.biAcceptor = biAcceptor;
+      return StartClient::new;
+    }
+
     public ClientRSocketFactory fragment(int mtu) {
       this.mtu = mtu;
       return this;
@@ -293,9 +302,27 @@ public Mono<RSocket> start() {
                           keepAliveTimeout(),
                           keepAliveHandler);
 
+                  ByteBuf setupFrame =
+                      SetupFrameFlyweight.encode(
+                          allocator,
+                          false,
+                          keepAliveTickPeriod(),
+                          keepAliveTimeout(),
+                          resumeToken,
+                          metadataMimeType,
+                          dataMimeType,
+                          setupPayload.sliceMetadata(),
+                          setupPayload.sliceData());
+
                   RSocket wrappedRSocketClient = plugins.applyClient(rSocketClient);
 
-                  RSocket unwrappedServerSocket = acceptor.get().apply(wrappedRSocketClient);
+                  RSocket unwrappedServerSocket;
+                  if (biAcceptor != null) {
+                    ConnectionSetupPayload setup = ConnectionSetupPayload.create(setupFrame);
+                    unwrappedServerSocket = biAcceptor.apply(setup, wrappedRSocketClient);
+                  } else {
+                    unwrappedServerSocket = acceptor.get().apply(wrappedRSocketClient);
+                  }
 
                   RSocket wrappedRSocketServer = plugins.applyServer(unwrappedServerSocket);
 
@@ -307,18 +334,6 @@ public Mono<RSocket> start() {
                           payloadDecoder,
                           errorConsumer);
 
-                  ByteBuf setupFrame =
-                      SetupFrameFlyweight.encode(
-                          allocator,
-                          false,
-                          keepAliveTickPeriod(),
-                          keepAliveTimeout(),
-                          resumeToken,
-                          metadataMimeType,
-                          dataMimeType,
-                          setupPayload.sliceMetadata(),
-                          setupPayload.sliceData());
-
                   return wrappedConnection.sendOne(setupFrame).thenReturn(wrappedRSocketClient);
                 });
       }