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