Propagate max frame length to WebSocket session

This change ensures that maxFramePayloadLength from
WebsocketClientSpec is passed to ReactorNettyWebSocketSession.

Previously, the session used the default 64KB limit regardless
of client configuration, causing TooLongFrameException when
receiving larger frames from servers like Tomcat or Jetty.

Closes gh-36369

Signed-off-by: Artem Voronin <artem.voronin.dev@gmail.com>

Author: aavoronin93

spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java

@@ -127,16 +127,18 @@ public Mono<Void> execute(URI url, WebSocketHandler handler) {
 	@Override
 	public Mono<Void> execute(URI url, HttpHeaders requestHeaders, WebSocketHandler handler) {
 		String protocols = StringUtils.collectionToCommaDelimitedString(handler.getSubProtocols());
+		WebsocketClientSpec wsClientSpec = buildSpec(protocols);
 		return getHttpClient()
 				.headers(nettyHeaders -> setNettyHeaders(requestHeaders, nettyHeaders))
-				.websocket(buildSpec(protocols))
+				.websocket(wsClientSpec)
 				.uri(url.toString())
 				.handle((inbound, outbound) -> {
 					HttpHeaders responseHeaders = toHttpHeaders(inbound);
 					String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol");
 					HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol);
 					NettyDataBufferFactory factory = new NettyDataBufferFactory(outbound.alloc());
-					WebSocketSession session = new ReactorNettyWebSocketSession(inbound, outbound, info, factory);
+					WebSocketSession session = new ReactorNettyWebSocketSession(inbound, outbound, info, factory,
+							wsClientSpec.maxFramePayloadLength());
 					if (logger.isDebugEnabled()) {
 						logger.debug("Started session '" + session.getId() + "' for " + url);
 					}

spring-webflux/src/test/java/org/springframework/web/reactive/socket/AbstractReactiveWebSocketIntegrationTests.java

@@ -148,7 +148,9 @@ void stopServer() {
 		if (this.client instanceof Lifecycle lifecycle) {
 			lifecycle.stop();
 		}
-		this.server.stop();
+		if (this.server != null) {
+			this.server.stop();
+		}
 	}
 
 

spring-webflux/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java

@@ -16,7 +16,9 @@
 
 package org.springframework.web.reactive.socket;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -27,21 +29,27 @@
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.netty.http.client.WebsocketClientSpec;
 import reactor.util.retry.Retry;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.ResponseCookie;
 import org.springframework.web.filter.reactive.ServerWebExchangeContextFilter;
 import org.springframework.web.reactive.HandlerMapping;
 import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
+import org.springframework.web.reactive.socket.adapter.NettyWebSocketSessionSupport;
+import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
 import org.springframework.web.reactive.socket.client.WebSocketClient;
 import org.springframework.web.server.WebFilter;
 import org.springframework.web.testfixture.http.server.reactive.bootstrap.HttpServer;
 import org.springframework.web.testfixture.http.server.reactive.bootstrap.TomcatHttpServer;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 /**
  * Integration tests with server-side {@link WebSocketHandler}s.
@@ -186,6 +194,35 @@ void cookie(WebSocketClient client, HttpServer server, Class<?> serverConfigClas
 		assertThat(cookie.get()).isEqualTo("project=spring");
 	}
 
+	@ParameterizedWebSocketTest
+	void largePayload(WebSocketClient client, HttpServer server, Class<?> serverConfigClass) throws Exception {
+
+		assumeTrue(client instanceof ReactorNettyWebSocketClient);
+
+		int defaultFrameMaxSize = NettyWebSocketSessionSupport.DEFAULT_FRAME_MAX_SIZE;
+		int extendedLimit = 2 * defaultFrameMaxSize;
+
+		ReactorNettyWebSocketClient nettyClient = (ReactorNettyWebSocketClient) client;
+		ReactorNettyWebSocketClient reconfiguredClient = new ReactorNettyWebSocketClient(
+				nettyClient.getHttpClient(),
+				() -> WebsocketClientSpec.builder().maxFramePayloadLength(extendedLimit));
+
+		startServer(reconfiguredClient, server, serverConfigClass);
+
+		AtomicReference<Integer> payloadSizeRef = new AtomicReference<>();
+		assertThatCode(() -> reconfiguredClient.execute(getUrl("/large-payload"),
+						session -> session.receive()
+								.map(WebSocketMessage::getPayload)
+								.map(DataBuffer::readableByteCount)
+								.reduce(Integer::sum)
+								.doOnNext(payloadSizeRef::set)
+								.then())
+				.block(TIMEOUT))
+				.doesNotThrowAnyException();
+
+		assertThat(payloadSizeRef.get()).isGreaterThan(defaultFrameMaxSize);
+		assertThat(payloadSizeRef.get()).isEqualTo(extendedLimit);
+	}
 
 	@Configuration
 	static class WebConfig {
@@ -198,6 +235,7 @@ public HandlerMapping handlerMapping() {
 			map.put("/custom-header", new CustomHeaderHandler());
 			map.put("/close", new SessionClosingHandler());
 			map.put("/cookie", new CookieHandler());
+			map.put("/large-payload", new LargePayloadHandler());
 			return new SimpleUrlHandlerMapping(map);
 		}
 
@@ -274,4 +312,15 @@ public Mono<Void> handle(WebSocketSession session) {
 		}
 	}
 
+	private static class LargePayloadHandler implements WebSocketHandler {
+
+		@Override
+		public Mono<Void> handle(WebSocketSession session) {
+			int doubledFrameSize = 2 * NettyWebSocketSessionSupport.DEFAULT_FRAME_MAX_SIZE;
+			byte[] payload = new byte[doubledFrameSize];
+			Arrays.fill(payload, (byte) 'x');
+			String text = new String(payload, StandardCharsets.UTF_8);
+			return session.send(Mono.just(session.textMessage(text)));
+		}
+	}
 }