From 11bdb43aadd6e3ccd86d18aa271c418a7b3eca4c Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Tue, 23 Jun 2026 10:12:37 +0100 Subject: [PATCH 1/2] Polishing in ConcurrentWebSocketSessionDecorator See gh-36909 --- .../ConcurrentWebSocketSessionDecorator.java | 33 ++++---- .../handler/BlockingWebSocketSession.java | 5 +- ...currentWebSocketSessionDecoratorTests.java | 79 ++++++++----------- 3 files changed, 51 insertions(+), 66 deletions(-) diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java index 0113f89b0bf..042f3e6c045 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java @@ -260,20 +260,7 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat if (this.closeInProgress) { return; } - if (!CloseStatus.SESSION_NOT_RELIABLE.equals(status)) { - try { - checkSessionLimits(); - } - catch (SessionLimitExceededException ex) { - // Ignore - } - if (this.limitExceeded) { - if (logger.isDebugEnabled()) { - logger.debug("Changing close status " + status + " to SESSION_NOT_RELIABLE."); - } - status = CloseStatus.SESSION_NOT_RELIABLE; - } - } + status = checkSessionLimitsAndChangeStatusIfNecessary(status); this.closeInProgress = true; super.close(status); } @@ -283,6 +270,24 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat } } + private CloseStatus checkSessionLimitsAndChangeStatusIfNecessary(CloseStatus status) { + if (!CloseStatus.SESSION_NOT_RELIABLE.equals(status)) { + try { + checkSessionLimits(); + } + catch (SessionLimitExceededException ex) { + // Ignore + } + if (this.limitExceeded) { + if (logger.isDebugEnabled()) { + logger.debug("Changing close status " + status + " to SESSION_NOT_RELIABLE."); + } + status = CloseStatus.SESSION_NOT_RELIABLE; + } + } + return status; + } + @Override public String toString() { diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/handler/BlockingWebSocketSession.java b/spring-websocket/src/test/java/org/springframework/web/socket/handler/BlockingWebSocketSession.java index 4398445c20b..18c57f104ed 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/handler/BlockingWebSocketSession.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/handler/BlockingWebSocketSession.java @@ -32,8 +32,6 @@ public class BlockingWebSocketSession extends TestWebSocketSession { private final AtomicReference sendLatch = new AtomicReference<>(); - private final AtomicReference releaseLatch = new AtomicReference<>(); - public CountDownLatch initSendLatch() { this.sendLatch.set(new CountDownLatch(1)); @@ -51,8 +49,7 @@ public class BlockingWebSocketSession extends TestWebSocketSession { private void block() { try { - this.releaseLatch.set(new CountDownLatch(1)); - this.releaseLatch.get().await(); + new CountDownLatch(1).await(); } catch (InterruptedException ex) { ex.printStackTrace(); diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java index f2466726bf7..746e5d2dae4 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java @@ -68,7 +68,7 @@ class ConcurrentWebSocketSessionDecoratorTests { sendBlockingMessage(decorator); - Thread.sleep(50); + Thread.sleep(5); assertThat(decorator.getTimeSinceSendStarted()).isGreaterThan(0); TextMessage payload = new TextMessage("payload"); @@ -89,18 +89,18 @@ class ConcurrentWebSocketSessionDecoratorTests { session.setOpen(true); ConcurrentWebSocketSessionDecorator decorator = - new ConcurrentWebSocketSessionDecorator(session, 100, 1024); + new ConcurrentWebSocketSessionDecorator(session, 50, 1024); sendBlockingMessage(decorator); // Exceed send time - Thread.sleep(200); + Thread.sleep(60); TextMessage payload = new TextMessage("payload"); - assertThatExceptionOfType(SessionLimitExceededException.class).isThrownBy(() -> - decorator.sendMessage(payload)) - .withMessageMatching("Send time [\\d]+ \\(ms\\) for session '123' exceeded the allowed limit 100") - .satisfies(ex -> assertThat(ex.getStatus()).isEqualTo(CloseStatus.SESSION_NOT_RELIABLE)); + assertThatExceptionOfType(SessionLimitExceededException.class) + .isThrownBy(() -> decorator.sendMessage(payload)) + .withMessageMatching("Send time [\\d]+ \\(ms\\) for session '123' exceeded the allowed limit 50") + .satisfies(ex -> assertThat(ex.getStatus()).isEqualTo(CloseStatus.SESSION_NOT_RELIABLE)); } @Test @@ -111,7 +111,7 @@ class ConcurrentWebSocketSessionDecoratorTests { session.setOpen(true); ConcurrentWebSocketSessionDecorator decorator = - new ConcurrentWebSocketSessionDecorator(session, 10*1000, 1024); + new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 1024); sendBlockingMessage(decorator); @@ -122,10 +122,10 @@ class ConcurrentWebSocketSessionDecoratorTests { assertThat(decorator.getBufferSize()).isEqualTo(1023); assertThat(session.isOpen()).isTrue(); - assertThatExceptionOfType(SessionLimitExceededException.class).isThrownBy(() -> - decorator.sendMessage(message)) - .withMessageMatching("Buffer size [\\d]+ bytes for session '123' exceeds the allowed limit 1024") - .satisfies(ex -> assertThat(ex.getStatus()).isEqualTo(CloseStatus.SESSION_NOT_RELIABLE)); + assertThatExceptionOfType(SessionLimitExceededException.class) + .isThrownBy(() -> decorator.sendMessage(message)) + .withMessageMatching("Buffer size [\\d]+ bytes for session '123' exceeds the allowed limit 1024") + .satisfies(ex -> assertThat(ex.getStatus()).isEqualTo(CloseStatus.SESSION_NOT_RELIABLE)); } @Test // SPR-17140 @@ -136,12 +136,11 @@ class ConcurrentWebSocketSessionDecoratorTests { session.setOpen(true); ConcurrentWebSocketSessionDecorator decorator = - new ConcurrentWebSocketSessionDecorator(session, 10*1000, 1024, OverflowStrategy.DROP); + new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 1024, OverflowStrategy.DROP); sendBlockingMessage(decorator); String msg = String.format("%1023s", "a"); - for (int i = 0; i < 5; i++) { TextMessage message = new TextMessage(msg); decorator.sendMessage(message); @@ -156,7 +155,7 @@ class ConcurrentWebSocketSessionDecoratorTests { BlockingWebSocketSession session = new BlockingWebSocketSession(); session.setOpen(true); - WebSocketSession decorator = new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 1024); + WebSocketSession decorator = new ConcurrentWebSocketSessionDecorator(session, 50, 1024); decorator.close(CloseStatus.PROTOCOL_ERROR); assertThat(session.getCloseStatus()).isEqualTo(CloseStatus.PROTOCOL_ERROR); @@ -171,28 +170,12 @@ class ConcurrentWebSocketSessionDecoratorTests { BlockingWebSocketSession session = new BlockingWebSocketSession(); session.setId("123"); session.setOpen(true); - CountDownLatch sentMessageLatch = session.initSendLatch(); - - int sendTimeLimit = 100; - int bufferSizeLimit = 1024; ConcurrentWebSocketSessionDecorator decorator = - new ConcurrentWebSocketSessionDecorator(session, sendTimeLimit, bufferSizeLimit); + new ConcurrentWebSocketSessionDecorator(session, 50, 1024); - Executors.newSingleThreadExecutor().submit(() -> { - TextMessage message = new TextMessage("slow message"); - try { - decorator.sendMessage(message); - } - catch (IOException e) { - e.printStackTrace(); - } - }); - - assertThat(sentMessageLatch.await(5, TimeUnit.SECONDS)).isTrue(); - - // ensure some send time elapses - Thread.sleep(sendTimeLimit + 100); + sendBlockingMessage(decorator); + Thread.sleep(55); decorator.close(CloseStatus.PROTOCOL_ERROR); @@ -201,20 +184,6 @@ class ConcurrentWebSocketSessionDecoratorTests { .isEqualTo(CloseStatus.SESSION_NOT_RELIABLE); } - private void sendBlockingMessage(ConcurrentWebSocketSessionDecorator session) throws InterruptedException { - CountDownLatch latch = ((BlockingWebSocketSession) session.getDelegate()).initSendLatch(); - Executors.newSingleThreadExecutor().submit(() -> { - TextMessage message = new TextMessage("slow message"); - try { - session.sendMessage(message); - } - catch (IOException e) { - e.printStackTrace(); - } - }); - assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); - } - @Test void configuredProperties() { TestWebSocketSession session = new TestWebSocketSession(); @@ -226,4 +195,18 @@ class ConcurrentWebSocketSessionDecoratorTests { assertThat(sessionDecorator.getOverflowStrategy()).isEqualTo(OverflowStrategy.DROP); } + private void sendBlockingMessage(ConcurrentWebSocketSessionDecorator session) throws InterruptedException { + CountDownLatch latch = ((BlockingWebSocketSession) session.getDelegate()).initSendLatch(); + Executors.newSingleThreadExecutor().submit(() -> { + TextMessage message = new TextMessage("slow message"); + try { + session.sendMessage(message); + } + catch (IOException ex) { + ex.printStackTrace(); + } + }); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + } + } From 6ac642e301c274b9c3c8968af0d38ee186219f6a Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Thu, 25 Jun 2026 12:50:04 +0100 Subject: [PATCH 2/2] Remove closeLock from ConcurrentWebSocketSessionDecorator#checkSessionLimits Closes gh-36909 --- .../ConcurrentWebSocketSessionDecorator.java | 65 +++++++++---------- 1 file changed, 30 insertions(+), 35 deletions(-) diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java index 042f3e6c045..c6b989434bd 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java @@ -208,42 +208,37 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat } private void checkSessionLimits() { - if (!shouldNotSend() && this.closeLock.tryLock()) { - try { - if (getTimeSinceSendStarted() > getSendTimeLimit()) { - String format = "Send time %d (ms) for session '%s' exceeded the allowed limit %d"; - String reason = String.format(format, getTimeSinceSendStarted(), getId(), getSendTimeLimit()); - limitExceeded(reason); - } - else if (getBufferSize() > getBufferSizeLimit()) { - switch (this.overflowStrategy) { - case TERMINATE -> { - String format = "Buffer size %d bytes for session '%s' exceeds the allowed limit %d"; - String reason = String.format(format, getBufferSize(), getId(), getBufferSizeLimit()); - limitExceeded(reason); - } - case DROP -> { - int i = 0; - while (getBufferSize() > getBufferSizeLimit()) { - WebSocketMessage message = this.buffer.poll(); - if (message == null) { - break; - } - this.bufferSize.addAndGet(-message.getPayloadLength()); - i++; - } - if (logger.isDebugEnabled()) { - logger.debug("Dropped " + i + " messages, buffer size: " + getBufferSize()); - } - } - default -> - // Should never happen.. - throw new IllegalStateException("Unexpected OverflowStrategy: " + this.overflowStrategy); - } - } + if (!shouldNotSend()) { + if (getTimeSinceSendStarted() > getSendTimeLimit()) { + String format = "Send time %d (ms) for session '%s' exceeded the allowed limit %d"; + String reason = String.format(format, getTimeSinceSendStarted(), getId(), getSendTimeLimit()); + limitExceeded(reason); } - finally { - this.closeLock.unlock(); + else if (getBufferSize() > getBufferSizeLimit()) { + switch (this.overflowStrategy) { + case TERMINATE -> { + String format = "Buffer size %d bytes for session '%s' exceeds the allowed limit %d"; + String reason = String.format(format, getBufferSize(), getId(), getBufferSizeLimit()); + limitExceeded(reason); + } + case DROP -> { + int i = 0; + while (getBufferSize() > getBufferSizeLimit()) { + WebSocketMessage message = this.buffer.poll(); + if (message == null) { + break; + } + this.bufferSize.addAndGet(-message.getPayloadLength()); + i++; + } + if (logger.isDebugEnabled()) { + logger.debug("Dropped " + i + " messages, buffer size: " + getBufferSize()); + } + } + default -> + // Should never happen.. + throw new IllegalStateException("Unexpected OverflowStrategy: " + this.overflowStrategy); + } } } }