Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch BufferingStompDecoder to decoration

Browse Source 
The BufferingStompDecoder now decorates rather than extend
StompDecoder. This allows a single StompDecoder instance to be
configured and extended independantly while buffering remains a
separate concern.
This commit is contained in:
Rossen Stoyanchev
2014-04-13 21:35:22 -04:00
parent 1369ff4d84
commit 65b17b80ba
3 changed files with 35 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java
View File

@@ -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<Message<byte[]>> decode(ByteBuffer newBuffer) {
this.chunks.add(newBuffer);
@@ -119,7 +130,7 @@ public class BufferingStompDecoder extends StompDecoder {
ByteBuffer bufferToDecode = assembleChunksAndReset();
MultiValueMap<String, String> headers = new LinkedMultiValueMap<String, String>();
List<Message<byte[]>> messages = decode(bufferToDecode, headers);
List<Message<byte[]>> messages = this.stompDecoder.decode(bufferToDecode, headers);
if (bufferToDecode.hasRemaining()) {
this.chunks.add(bufferToDecode);

27
spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/BufferingStompDecoderTests.java
View File

@@ -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<Message<byte[]>> 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<Message<byte[]>> messages = stompDecoder.decode(toByteBuffer(chunk1));
assertEquals(Arrays.asList(), messages);
assertEquals(Collections.<Message<byte[]>>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<Message<byte[]>> 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<Message<byte[]>> 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<Message<byte[]>> 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<Message<byte[]>> 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<String, String> headers = new LinkedMultiValueMap<>();
List<Message<byte[]>> messages = stompDecoder.decode(toByteBuffer(chunk), headers);
List<Message<byte[]>> messages = stompDecoder.decode(toByteBuffer(chunk));
assertEquals(0, messages.size());
}

11
spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java
View File

@@ -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<String, BufferingStompDecoder> decoders = new ConcurrentHashMap<String, BufferingStompDecoder>();
private final StompEncoder stompEncoder = new StompEncoder();
private final StompDecoder stompDecoder = new StompDecoder();
private final Map<String, BufferingStompDecoder> decoders = new ConcurrentHashMap<String, BufferingStompDecoder>();
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<Message<byte[]>> messages = null;
List<Message<byte[]>> 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