diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java index ba9afe0079..17728a7577 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java @@ -46,7 +46,9 @@ import java.util.concurrent.LinkedBlockingQueue; * @author Rossen Stoyanchev * @since 4.0.3 */ -public class BufferingStompDecoder extends StompDecoder { +public class BufferingStompDecoder { + + private final StompDecoder stompDecoder; private final int bufferSizeLimit; @@ -55,12 +57,22 @@ public class BufferingStompDecoder extends StompDecoder { private volatile Integer expectedContentLength; - public BufferingStompDecoder(int bufferSizeLimit) { + public BufferingStompDecoder(StompDecoder stompDecoder, int bufferSizeLimit) { + Assert.notNull(stompDecoder, "'stompDecoder' is required"); Assert.isTrue(bufferSizeLimit > 0, "Buffer size must be greater than 0"); + this.stompDecoder = stompDecoder; this.bufferSizeLimit = bufferSizeLimit; } + /** + * Return the wrapped + * {@link org.springframework.messaging.simp.stomp.StompDecoder}. + */ + public StompDecoder getStompDecoder() { + return this.stompDecoder; + } + /** * Return the configured buffer size limit. */ @@ -105,7 +117,6 @@ public class BufferingStompDecoder extends StompDecoder { * @return decoded messages or an empty list * @throws StompConversionException raised in case of decoding issues */ - @Override public List> decode(ByteBuffer newBuffer) { this.chunks.add(newBuffer); @@ -119,7 +130,7 @@ public class BufferingStompDecoder extends StompDecoder { ByteBuffer bufferToDecode = assembleChunksAndReset(); MultiValueMap headers = new LinkedMultiValueMap(); - List> messages = decode(bufferToDecode, headers); + List> messages = this.stompDecoder.decode(bufferToDecode, headers); if (bufferToDecode.hasRemaining()) { this.chunks.add(bufferToDecode); diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/BufferingStompDecoderTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/BufferingStompDecoderTests.java index 35d7dd03c8..a00d4ea323 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/BufferingStompDecoderTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/BufferingStompDecoderTests.java @@ -18,12 +18,10 @@ package org.springframework.messaging.simp.stomp; import org.junit.Test; import org.springframework.messaging.Message; -import org.springframework.messaging.converter.MessageConversionException; -import org.springframework.util.LinkedMultiValueMap; import java.nio.ByteBuffer; import java.nio.charset.Charset; -import java.util.Arrays; +import java.util.Collections; import java.util.List; import static org.junit.Assert.assertEquals; @@ -39,11 +37,12 @@ import static org.junit.Assert.fail; */ public class BufferingStompDecoderTests { + private final StompDecoder STOMP_DECODER = new StompDecoder(); @Test public void basic() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk = "SEND\na:alpha\n\nMessage body\0"; List> messages = stompDecoder.decode(toByteBuffer(chunk)); @@ -57,12 +56,12 @@ public class BufferingStompDecoderTests { @Test public void oneMessageInTwoChunks() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk1 = "SEND\na:alpha\n\nMessage"; String chunk2 = " body\0"; List> messages = stompDecoder.decode(toByteBuffer(chunk1)); - assertEquals(Arrays.asList(), messages); + assertEquals(Collections.>emptyList(), messages); messages = stompDecoder.decode(toByteBuffer(chunk2)); assertEquals(1, messages.size()); @@ -75,7 +74,7 @@ public class BufferingStompDecoderTests { @Test public void twoMessagesInOneChunk() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk = "SEND\na:alpha\n\nPayload1\0" + "SEND\na:alpha\n\nPayload2\0"; List> messages = stompDecoder.decode(toByteBuffer(chunk)); @@ -92,7 +91,7 @@ public class BufferingStompDecoderTests { int contentLength = "Payload2a-Payload2b".getBytes().length; - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk1 = "SEND\na:alpha\n\nPayload1\0SEND\ncontent-length:" + contentLength + "\n"; List> messages = stompDecoder.decode(toByteBuffer(chunk1)); @@ -121,7 +120,7 @@ public class BufferingStompDecoderTests { @Test public void oneFullAndOneSplitMessageNoContentLength() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk1 = "SEND\na:alpha\n\nPayload1\0SEND\na:alpha\n"; List> messages = stompDecoder.decode(toByteBuffer(chunk1)); @@ -150,7 +149,7 @@ public class BufferingStompDecoderTests { @Test public void oneFullAndOneSplitWithContentLengthExceedingBufferSize() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk1 = "SEND\na:alpha\n\nPayload1\0SEND\ncontent-length:129\n"; List> messages = stompDecoder.decode(toByteBuffer(chunk1)); @@ -166,12 +165,13 @@ public class BufferingStompDecoderTests { fail("Expected exception"); } catch (StompConversionException ex) { + // expected } } @Test(expected = StompConversionException.class) public void bufferSizeLimit() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(10); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 10); String payload = "SEND\na:alpha\n\nMessage body"; stompDecoder.decode(toByteBuffer(payload)); } @@ -179,11 +179,10 @@ public class BufferingStompDecoderTests { @Test public void incompleteCommand() throws InterruptedException { - BufferingStompDecoder stompDecoder = new BufferingStompDecoder(128); + BufferingStompDecoder stompDecoder = new BufferingStompDecoder(STOMP_DECODER, 128); String chunk = "MESSAG"; - LinkedMultiValueMap headers = new LinkedMultiValueMap<>(); - List> messages = stompDecoder.decode(toByteBuffer(chunk), headers); + List> messages = stompDecoder.decode(toByteBuffer(chunk)); assertEquals(0, messages.size()); } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java index 24413dfa0b..9cb3ae8384 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java @@ -38,6 +38,7 @@ import org.springframework.messaging.simp.SimpMessageType; import org.springframework.messaging.simp.stomp.BufferingStompDecoder; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.simp.stomp.StompConversionException; +import org.springframework.messaging.simp.stomp.StompDecoder; import org.springframework.messaging.simp.stomp.StompEncoder; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; import org.springframework.messaging.simp.user.DestinationUserNameProvider; @@ -84,10 +85,12 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE private int messageSizeLimit = 64 * 1024; - private final Map decoders = new ConcurrentHashMap(); - private final StompEncoder stompEncoder = new StompEncoder(); + private final StompDecoder stompDecoder = new StompDecoder(); + + private final Map decoders = new ConcurrentHashMap(); + private UserSessionRegistry userSessionRegistry; private ApplicationEventPublisher eventPublisher; @@ -148,7 +151,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE public void handleMessageFromClient(WebSocketSession session, WebSocketMessage webSocketMessage, MessageChannel outputChannel) { - List> messages = null; + List> messages; try { Assert.isInstanceOf(TextMessage.class, webSocketMessage); TextMessage textMessage = (TextMessage) webSocketMessage; @@ -380,7 +383,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE if (session.getTextMessageSizeLimit() < MINIMUM_WEBSOCKET_MESSAGE_SIZE) { session.setTextMessageSizeLimit(MINIMUM_WEBSOCKET_MESSAGE_SIZE); } - this.decoders.put(session.getId(), new BufferingStompDecoder(getMessageSizeLimit())); + this.decoders.put(session.getId(), new BufferingStompDecoder(this.stompDecoder, getMessageSizeLimit())); } @Override