Client SocketAcceptor can't access ConnectionSetupPayload if compression is on

[rstoyanchev]

If compression is enabled on client and server, the client acceptor fails with ArrayIndexOutOfBoundsException if it tries to access properties of ConnectionSetupPayload.

Here is the diff with changes to the WebSocketAggregationSample to demonstrate the issue:

diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/ws/WebSocketAggregationSample.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/ws/WebSocketAggregationSample.java
index 89304853..59a54c39 100644
--- a/rsocket-examples/src/main/java/io/rsocket/examples/transport/ws/WebSocketAggregationSample.java
+++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/ws/WebSocketAggregationSample.java
@@ -33,6 +33,7 @@
 import reactor.netty.Connection;
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
+import reactor.netty.http.server.WebsocketServerSpec;
 
 public class WebSocketAggregationSample {
 
@@ -57,16 +58,27 @@ public static void main(String[] args) {
                                 .apply(
                                     new WebsocketDuplexConnection(
                                         (Connection) in.aggregateFrames()))
-                                .then(out.neverComplete())))
+                                .then(out.neverComplete()),
+                            WebsocketServerSpec.builder().compress(true).build()))
             .bindNow();
 
     WebsocketClientTransport transport =
-        WebsocketClientTransport.create(server.host(), server.port());
+        WebsocketClientTransport.create(server.host(), server.port())
+                .webSocketSpec(builder -> builder.compress(true));
 
     RSocket clientRSocket =
         RSocketConnector.create()
             .keepAlive(Duration.ofMinutes(10), Duration.ofMinutes(10))
             .payloadDecoder(PayloadDecoder.ZERO_COPY)
+            .acceptor((setup, sendingSocket) -> {
+                try {
+                    setup.dataMimeType();
+                }
+                catch (Exception ex) {
+                    ex.printStackTrace();
+                }
+              return Mono.just(new RSocket() {});
+            })
             .connect(transport)
             .block();

I have noticed that the server acceptor has no trouble reading the ConnectionSetupPayload so it must be something client-side related, and it could be deeper in Reactor Netty.

This was originally reported in https://github.com/spring-projects/spring-framework/issues/27973 by @ekuleshov but the issue can be demonstrated without Spring.