Skip to content

Fix ByteBuffer concurrency issue in ConcurrentWebSocketSessionDecorator for binary messages #35519

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

Fix ByteBuffer concurrency issue in ConcurrentWebSocketSessionDecorator for binary messages #35519

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d2af44
Add test to reproduce ByteBuffer concurrency issue in WebSocket sessions
xeromank Sep 22, 2025
23300bc
Fix thread-safety issue with shared BinaryMessage in concurrent sessions
xeromank Sep 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 14 additions & 3 deletions ...main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.springframework.web.socket.handler;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
Expand All @@ -28,6 +29,7 @@
import org.apache.commons.logging.LogFactory;
import org.jspecify.annotations.Nullable;

import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
Expand All @@ -43,6 +45,7 @@
*
* @author Rossen Stoyanchev
* @author Juergen Hoeller
* @author xeroman.k
* @since 4.0.3
*/
public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorator {
Expand Down Expand Up @@ -170,7 +173,7 @@ public void sendMessage(WebSocketMessage<?> message) throws IOException {
if (!tryFlushMessageBuffer()) {
if (logger.isTraceEnabled()) {
logger.trace(String.format("Another send already in progress: " +
"session id '%s':, \"in-progress\" send time %d (ms), buffer size %d bytes",
"session id '%s':, \"in-progress\" send time %d (ms), buffer size %d bytes",
getId(), getTimeSinceSendStarted(), getBufferSize()));
}
checkSessionLimits();
Expand All @@ -194,7 +197,7 @@ private boolean tryFlushMessageBuffer() throws IOException {
}
this.bufferSize.addAndGet(-message.getPayloadLength());
this.sendStartTime = System.currentTimeMillis();
getDelegate().sendMessage(message);
getDelegate().sendMessage(prepareMessage(message));
this.sendStartTime = 0;
}
}
Expand All @@ -207,6 +210,14 @@ private boolean tryFlushMessageBuffer() throws IOException {
return false;
}

private WebSocketMessage<?> prepareMessage(WebSocketMessage<?> message) {
if (message instanceof BinaryMessage) {
ByteBuffer payload = ((BinaryMessage) message).getPayload();
return new BinaryMessage(payload.duplicate(), message.isLast());
}
return message;
}

private void checkSessionLimits() {
if (!shouldNotSend() && this.closeLock.tryLock()) {
try {
Expand Down Expand Up @@ -238,7 +249,7 @@ else if (getBufferSize() > getBufferSizeLimit()) {
}
default ->
// Should never happen..
throw new IllegalStateException("Unexpected OverflowStrategy: " + this.overflowStrategy);
throw new IllegalStateException("Unexpected OverflowStrategy: " + this.overflowStrategy);
}
}
}
Expand Down
135 changes: 129 additions & 6 deletions ...java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,21 @@
package org.springframework.web.socket.handler;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.jupiter.api.Test;

import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.OverflowStrategy;

Expand All @@ -35,6 +42,7 @@
* Tests for {@link ConcurrentWebSocketSessionDecorator}.
*
* @author Rossen Stoyanchev
* @author xeroman.k
*/
class ConcurrentWebSocketSessionDecoratorTests {

Expand Down Expand Up @@ -98,9 +106,9 @@ void sendTimeLimitExceeded() throws InterruptedException {

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));
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));
}

@Test
Expand All @@ -123,9 +131,9 @@ void sendBufferSizeExceeded() throws IOException, InterruptedException {
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));
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
Expand Down Expand Up @@ -226,4 +234,119 @@ void configuredProperties() {
assertThat(sessionDecorator.getOverflowStrategy()).isEqualTo(OverflowStrategy.DROP);
}

@Test
void concurrentBinaryMessageSharingAcrossSessions() throws Exception {
byte[] originalData = new byte[100];
for (int i = 0; i < originalData.length; i++) {
originalData[i] = (byte) i;
}
ByteBuffer buffer = ByteBuffer.wrap(originalData);
BinaryMessage sharedMessage = new BinaryMessage(buffer);

int sessionCount = 5;
int messagesPerSession = 3;
CountDownLatch startLatch = new CountDownLatch(1);
CountDownLatch completeLatch = new CountDownLatch(sessionCount * messagesPerSession);
AtomicInteger corruptedBuffers = new AtomicInteger(0);

List<TestBinaryMessageCapturingSession> sessions = new ArrayList<>();
List<ConcurrentWebSocketSessionDecorator> decorators = new ArrayList<>();

for (int i = 0; i < sessionCount; i++) {
TestBinaryMessageCapturingSession session = new TestBinaryMessageCapturingSession();
session.setOpen(true);
session.setId("session-" + i);
sessions.add(session);

ConcurrentWebSocketSessionDecorator decorator =
new ConcurrentWebSocketSessionDecorator(session, 10000, 10240);
decorators.add(decorator);
}

ExecutorService executor = Executors.newFixedThreadPool(sessionCount * messagesPerSession);

try {
for (ConcurrentWebSocketSessionDecorator decorator : decorators) {
for (int j = 0; j < messagesPerSession; j++) {
executor.submit(() -> {
try {
startLatch.await();
decorator.sendMessage(sharedMessage);
} catch (Exception e) {
e.printStackTrace();
} finally {
completeLatch.countDown();
}
});
}
}

startLatch.countDown();
assertThat(completeLatch.await(5, TimeUnit.SECONDS)).isTrue();

for (TestBinaryMessageCapturingSession session : sessions) {
List<ByteBuffer> capturedBuffers = session.getCapturedBuffers();

for (ByteBuffer capturedBuffer : capturedBuffers) {
byte[] capturedData = new byte[capturedBuffer.remaining()];
capturedBuffer.get(capturedData);

boolean isCorrupted = false;
if (capturedData.length != originalData.length) {
isCorrupted = true;
} else {
for (int j = 0; j < originalData.length; j++) {
if (capturedData[j] != originalData[j]) {
isCorrupted = true;
break;
}
}
}

if (isCorrupted) {
corruptedBuffers.incrementAndGet();
}
}
}

assertThat(corruptedBuffers.get())
.as("No ByteBuffer corruption should occur with duplicate() fix")
.isEqualTo(0);
} finally {
executor.shutdown();
}
}

static class TestBinaryMessageCapturingSession extends TestWebSocketSession {
private final List<ByteBuffer> capturedBuffers = new ArrayList<>();

@Override
public void sendMessage(WebSocketMessage<?> message) throws IOException {
if (message instanceof BinaryMessage) {
ByteBuffer payload = ((BinaryMessage) message).getPayload();
ByteBuffer captured = ByteBuffer.allocate(payload.remaining());

while (payload.hasRemaining()) {
captured.put(payload.get());
}
captured.flip();

synchronized (capturedBuffers) {
capturedBuffers.add(captured);
}

try {
Thread.sleep(1);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
super.sendMessage(message);
}

public synchronized List<ByteBuffer> getCapturedBuffers() {
return new ArrayList<>(capturedBuffers);
}
}

}