spring/spring-framework
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-framework.git synced 2026-09-17 16:30:28 +00:00
Code Issues Projects Releases Wiki Activity

Polishing in ConcurrentWebSocketSessionDecorator

Browse Source 
See gh-36909
This commit is contained in:
rstoyanchev
2026-06-25 12:50:32 +01:00
parent 4074155d76
commit 11bdb43aad
3 changed files with 51 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java
+19 -14
View File
@@ -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() {
spring-websocket/src/test/java/org/springframework/web/socket/handler/BlockingWebSocketSession.java
+1 -4
View File
@@ -32,8 +32,6 @@ public class BlockingWebSocketSession extends TestWebSocketSession {
private final AtomicReference<CountDownLatch> sendLatch = new AtomicReference<>();
private final AtomicReference<CountDownLatch> 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();
spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java
+31 -48
View File
@@ -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();
}
}