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

Merge branch '7.0.x'

Browse Source
This commit is contained in:
rstoyanchev
2026-06-25 12:50:49 +01:00
parent 7e9da44e18 6ac642e301
commit ed0919b92e
3 changed files with 81 additions and 101 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
+49 -49
View File
@@ -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);
}
}
}
}
@@ -260,20 +255,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 +265,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();
}
}