spring-projectsGH-8678: Make buffer overflow strategy configurable in IntegrationWebSocketContainer

Author: ColoredCarrot

spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java

@@ -28,6 +28,7 @@
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.web.socket.CloseStatus;
@@ -67,6 +68,9 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
 
 	public static final int DEFAULT_SEND_BUFFER_SIZE = 512 * 1024;
 
+	public static final ConcurrentWebSocketSessionDecorator.OverflowStrategy DEFAULT_SEND_BUFFER_OVERFLOW_STRATEGY =
+			ConcurrentWebSocketSessionDecorator.OverflowStrategy.TERMINATE;
+
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
 
 	protected final Lock lock = new ReentrantLock();
@@ -83,6 +87,10 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
 
 	private int sendBufferSizeLimit = DEFAULT_SEND_BUFFER_SIZE;
 
+	@NonNull
+	private ConcurrentWebSocketSessionDecorator.OverflowStrategy sendBufferOverflowStrategy =
+			DEFAULT_SEND_BUFFER_OVERFLOW_STRATEGY;
+
 	public void setSendTimeLimit(int sendTimeLimit) {
 		this.sendTimeLimit = sendTimeLimit;
 	}
@@ -91,6 +99,24 @@ public void setSendBufferSizeLimit(int sendBufferSizeLimit) {
 		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}.
+	 * <p>
+	 * By default, the session associated with the culpable message is terminated.
+	 *
+	 * @param overflowStrategy The overflow strategy to use (see {@link ConcurrentWebSocketSessionDecorator.OverflowStrategy}).
+	 *
+	 * @see ConcurrentWebSocketSessionDecorator
+	 */
+	public void setSendBufferOverflowStrategy(@NonNull ConcurrentWebSocketSessionDecorator.OverflowStrategy overflowStrategy) {
+		Assert.notNull(overflowStrategy, "Overflow strategy must not be null");
+		this.sendBufferOverflowStrategy = overflowStrategy;
+	}
+
 	public void setMessageListener(WebSocketListener messageListener) {
 		Assert.state(this.messageListener == null || this.messageListener.equals(messageListener),
 				"'messageListener' is already configured");
@@ -190,7 +216,8 @@ public void afterConnectionEstablished(WebSocketSession sessionToDecorate)
 			WebSocketSession session =
 					new ConcurrentWebSocketSessionDecorator(sessionToDecorate,
 							IntegrationWebSocketContainer.this.sendTimeLimit,
-							IntegrationWebSocketContainer.this.sendBufferSizeLimit);
+							IntegrationWebSocketContainer.this.sendBufferSizeLimit,
+							IntegrationWebSocketContainer.this.sendBufferOverflowStrategy);
 
 			IntegrationWebSocketContainer.this.sessions.put(session.getId(), session);
 			if (IntegrationWebSocketContainer.this.logger.isDebugEnabled()) {