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`**

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.
+	 * <p>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()) {

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<String, Object> 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