From be53593af9fa745e5dc563abca2cd003173269c1 Mon Sep 17 00:00:00 2001 From: Julian Koch <7807570+ColoredCarrot@users.noreply.github.com> Date: Sun, 16 Jul 2023 13:30:45 -0400 Subject: [PATCH] GH-8678: Add BufferOverflowStrategy for WebScoket Within Spring Integration's WebSocket support, a `ConcurrentWebSocketSessionDecorator`, which buffers outbound messages if sending is slow, is used to decorate all websocket sessions in `IntegrationWebSocketContainer`, the standard entrypoint for using websockets with Integration. * Expose a `ConcurrentWebSocketSessionDecorator.OverflowStrategy` option on the `IntegrationWebSocketContainer` **Cherry-pick to `5.5.x`, `6.0.x` & `6.1.x`** --- .../IntegrationWebSocketContainer.java | 36 ++++++++++++-- .../ClientWebSocketContainerTests.java | 49 +++++++++++++++++++ 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java index dfe8d67e9c..1fad3076e8 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java @@ -28,6 +28,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; import org.springframework.web.socket.CloseStatus; @@ -55,6 +56,7 @@ import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorato * * @author Artem Bilan * @author Gary Russell + * @author Julian Koch * * @since 4.1 * @@ -83,6 +85,9 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { private int sendBufferSizeLimit = DEFAULT_SEND_BUFFER_SIZE; + @Nullable + private ConcurrentWebSocketSessionDecorator.OverflowStrategy sendBufferOverflowStrategy; + public void setSendTimeLimit(int sendTimeLimit) { this.sendTimeLimit = sendTimeLimit; } @@ -91,6 +96,21 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { this.sendBufferSizeLimit = sendBufferSizeLimit; } + /** + * Set the send buffer overflow strategy. + *

Concurrently generated outbound messages are buffered if sending is slow. + * This strategy determines the behavior when the buffer has reached the limit + * configured with {@link #setSendBufferSizeLimit}. + * @param overflowStrategy The {@link ConcurrentWebSocketSessionDecorator.OverflowStrategy} to use. + * @since 5.5.19 + * @see ConcurrentWebSocketSessionDecorator + */ + public void setSendBufferOverflowStrategy( + @Nullable ConcurrentWebSocketSessionDecorator.OverflowStrategy overflowStrategy) { + + this.sendBufferOverflowStrategy = overflowStrategy; + } + public void setMessageListener(WebSocketListener messageListener) { Assert.state(this.messageListener == null || this.messageListener.equals(messageListener), "'messageListener' is already configured"); @@ -166,6 +186,17 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { } } + private WebSocketSession decorateSession(WebSocketSession sessionToDecorate) { + if (this.sendBufferOverflowStrategy == null) { + return new ConcurrentWebSocketSessionDecorator(sessionToDecorate, this.sendTimeLimit, + this.sendBufferSizeLimit); + } + else { + return new ConcurrentWebSocketSessionDecorator(sessionToDecorate, this.sendTimeLimit, + this.sendBufferSizeLimit, this.sendBufferOverflowStrategy); + } + } + /** * An internal {@link WebSocketHandler} implementation to be used with native * Web-Socket containers. @@ -187,10 +218,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { public void afterConnectionEstablished(WebSocketSession sessionToDecorate) throws Exception { // NOSONAR - WebSocketSession session = - new ConcurrentWebSocketSessionDecorator(sessionToDecorate, - IntegrationWebSocketContainer.this.sendTimeLimit, - IntegrationWebSocketContainer.this.sendBufferSizeLimit); + WebSocketSession session = decorateSession(sessionToDecorate); IntegrationWebSocketContainer.this.sessions.put(session.getId(), session); if (IntegrationWebSocketContainer.this.logger.isDebugEnabled()) { diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java index 211abc0918..084962e9ab 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java @@ -34,6 +34,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.http.HttpHeaders; +import org.springframework.integration.test.util.TestUtils; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.PingMessage; import org.springframework.web.socket.PongMessage; @@ -42,12 +43,14 @@ import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.WebSocketMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; /** * @author Artem Bilan + * @author Julian Koch * * @since 4.1 */ @@ -138,16 +141,53 @@ public class ClientWebSocketContainerTests { assertThat(session.isOpen()).isTrue(); } + @Test + public void testWebSocketContainerOverflowStrategyPropagation() throws Exception { + StandardWebSocketClient webSocketClient = new StandardWebSocketClient(); + + Map userProperties = new HashMap<>(); + userProperties.put(Constants.IO_TIMEOUT_MS_PROPERTY, "" + (Constants.IO_TIMEOUT_MS_DEFAULT * 6)); + webSocketClient.setUserProperties(userProperties); + + ClientWebSocketContainer container = + new ClientWebSocketContainer(webSocketClient, new URI(server.getWsBaseUrl() + "/ws/websocket")); + + container.setSendTimeLimit(10_000); + container.setSendBufferSizeLimit(12345); + container.setSendBufferOverflowStrategy(ConcurrentWebSocketSessionDecorator.OverflowStrategy.DROP); + + TestWebSocketListener messageListener = new TestWebSocketListener(); + container.setMessageListener(messageListener); + container.setConnectionTimeout(30); + + container.start(); + + assertThat(messageListener.sessionStartedLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(messageListener.sendTimeLimit).isEqualTo(10_000); + assertThat(messageListener.sendBufferSizeLimit).isEqualTo(12345); + assertThat(messageListener.sendBufferOverflowStrategy) + .isEqualTo(ConcurrentWebSocketSessionDecorator.OverflowStrategy.DROP); + } + private static class TestWebSocketListener implements WebSocketListener { public final CountDownLatch messageLatch = new CountDownLatch(1); + public final CountDownLatch sessionStartedLatch = new CountDownLatch(1); + public final CountDownLatch sessionEndedLatch = new CountDownLatch(1); public WebSocketMessage message; public boolean started; + int sendTimeLimit; + + int sendBufferSizeLimit; + + ConcurrentWebSocketSessionDecorator.OverflowStrategy sendBufferOverflowStrategy; + TestWebSocketListener() { } @@ -160,6 +200,15 @@ public class ClientWebSocketContainerTests { @Override public void afterSessionStarted(WebSocketSession session) { this.started = true; + + var sessionDecorator = (ConcurrentWebSocketSessionDecorator) session; + this.sendTimeLimit = sessionDecorator.getSendTimeLimit(); + this.sendBufferSizeLimit = sessionDecorator.getBufferSizeLimit(); + this.sendBufferOverflowStrategy = + TestUtils.getPropertyValue(sessionDecorator, "overflowStrategy", + ConcurrentWebSocketSessionDecorator.OverflowStrategy.class); + + this.sessionStartedLatch.countDown(); } @Override