Add StompCodec
Previously, the broker relay's TCP client used Reactor's built in delimited codec as part of its parsing of STOMP frames. \0 was used as the delimiter. This worked for most STOMP frames but, crucially, not for frames with a body that contained \0: when such a frame was received it would be truncated. This commit adds a custom codec that parses STOMP frames more intelligently. It honours the content-length header allowing it to correctly parse frames with a body that contains \0. The codec largely delegates to two new classes: StompEncoder and StompDecoder. For consistency, code that previously used StompMessageConverter has been reworked to use these new encoder and decoder classes. Issue: SPR-10818
This commit is contained in:
Andy Wilkinson
committed by
Rossen Stoyanchev
Rossen Stoyanchev
parent
f705ec1a46
commit
a489c2cf38
@@ -253,7 +253,7 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
|
||||
}
|
||||
|
||||
public void awaitAndAssert() throws InterruptedException {
|
||||
boolean result = this.latch.await(5000, TimeUnit.MILLISECONDS);
|
||||
boolean result = this.latch.await(10000, TimeUnit.MILLISECONDS);
|
||||
assertTrue(getAsString(), result && this.unexpected.isEmpty());
|
||||
}
|
||||
|
||||
@@ -356,6 +356,7 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
|
||||
public static MessageExchangeBuilder connect(String sessionId) {
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
|
||||
headers.setSessionId(sessionId);
|
||||
headers.setAcceptVersion("1.1,1.2");
|
||||
Message<?> message = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
|
||||
return new MessageExchangeBuilder(message);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
/*
|
||||
* Copyright 2002-2013 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.messaging.simp.stomp;
|
||||
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
|
||||
import reactor.function.Consumer;
|
||||
import reactor.function.Function;
|
||||
import reactor.io.Buffer;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* @author awilkinson
|
||||
*/
|
||||
public class StompCodecTests {
|
||||
|
||||
private final ArgumentCapturingConsumer<Message<byte[]>> consumer = new ArgumentCapturingConsumer<Message<byte[]>>();
|
||||
|
||||
private final Function<Buffer, Message<byte[]>> decoder = new StompCodec().decoder(consumer);
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithCrLfEols() {
|
||||
Message<byte[]> frame = decode("DISCONNECT\r\n\r\n\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.DISCONNECT, headers.getCommand());
|
||||
assertEquals(0, headers.toStompHeaderMap().size());
|
||||
assertEquals(0, frame.getPayload().length);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithNoHeadersAndNoBody() {
|
||||
Message<byte[]> frame = decode("DISCONNECT\n\n\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.DISCONNECT, headers.getCommand());
|
||||
assertEquals(0, headers.toStompHeaderMap().size());
|
||||
assertEquals(0, frame.getPayload().length);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithNoBody() {
|
||||
String accept = "accept-version:1.1\n";
|
||||
String host = "host:github.org\n";
|
||||
|
||||
Message<byte[]> frame = decode("CONNECT\n" + accept + host + "\n\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.CONNECT, headers.getCommand());
|
||||
|
||||
assertEquals(2, headers.toStompHeaderMap().size());
|
||||
assertEquals("1.1", headers.getFirstNativeHeader("accept-version"));
|
||||
assertEquals("github.org", headers.getHost());
|
||||
|
||||
assertEquals(0, frame.getPayload().length);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrame() throws UnsupportedEncodingException {
|
||||
Message<byte[]> frame = decode("SEND\ndestination:test\n\nThe body of the message\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.SEND, headers.getCommand());
|
||||
|
||||
assertEquals(1, headers.toStompHeaderMap().size());
|
||||
assertEquals("test", headers.getDestination());
|
||||
|
||||
String bodyText = new String(frame.getPayload());
|
||||
assertEquals("The body of the message", bodyText);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithContentLength() {
|
||||
Message<byte[]> frame = decode("SEND\ncontent-length:23\n\nThe body of the message\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.SEND, headers.getCommand());
|
||||
|
||||
assertEquals(1, headers.toStompHeaderMap().size());
|
||||
assertEquals(Integer.valueOf(23), headers.getContentLength());
|
||||
|
||||
String bodyText = new String(frame.getPayload());
|
||||
assertEquals("The body of the message", bodyText);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithNullOctectsInTheBody() {
|
||||
Message<byte[]> frame = decode("SEND\ncontent-length:23\n\nThe b\0dy \0f the message\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.SEND, headers.getCommand());
|
||||
|
||||
assertEquals(1, headers.toStompHeaderMap().size());
|
||||
assertEquals(Integer.valueOf(23), headers.getContentLength());
|
||||
|
||||
String bodyText = new String(frame.getPayload());
|
||||
assertEquals("The b\0dy \0f the message", bodyText);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeFrameWithEscapedHeaders() {
|
||||
Message<byte[]> frame = decode("DISCONNECT\na\\c\\r\\n\\\\b:alpha\\cbravo\\r\\n\\\\\n\n\0");
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.wrap(frame);
|
||||
|
||||
assertEquals(StompCommand.DISCONNECT, headers.getCommand());
|
||||
|
||||
assertEquals(1, headers.toStompHeaderMap().size());
|
||||
assertEquals("alpha:bravo\r\n\\", headers.getFirstNativeHeader("a:\r\n\\b"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void decodeMultipleFramesFromSameBuffer() {
|
||||
String frame1 = "SEND\ndestination:test\n\nThe body of the message\0";
|
||||
String frame2 = "DISCONNECT\n\n\0";
|
||||
|
||||
Buffer buffer = Buffer.wrap(frame1 + frame2);
|
||||
|
||||
final List<Message<byte[]>> messages = new ArrayList<Message<byte[]>>();
|
||||
new StompCodec().decoder(new Consumer<Message<byte[]>>() {
|
||||
@Override
|
||||
public void accept(Message<byte[]> message) {
|
||||
messages.add(message);
|
||||
}
|
||||
}).apply(buffer);
|
||||
|
||||
assertEquals(2, messages.size());
|
||||
assertEquals(StompCommand.SEND, StompHeaderAccessor.wrap(messages.get(0)).getCommand());
|
||||
assertEquals(StompCommand.DISCONNECT, StompHeaderAccessor.wrap(messages.get(1)).getCommand());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void encodeFrameWithNoHeadersAndNoBody() {
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.DISCONNECT);
|
||||
|
||||
Message<byte[]> frame = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
|
||||
|
||||
assertEquals("DISCONNECT\n\n\0", new StompCodec().encoder().apply(frame).asString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void encodeFrameWithHeaders() {
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
|
||||
headers.setAcceptVersion("1.2");
|
||||
headers.setHost("github.org");
|
||||
|
||||
Message<byte[]> frame = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
|
||||
|
||||
String frameString = new StompCodec().encoder().apply(frame).asString();
|
||||
|
||||
assertTrue(frameString.equals("CONNECT\naccept-version:1.2\nhost:github.org\n\n\0") ||
|
||||
frameString.equals("CONNECT\nhost:github.org\naccept-version:1.2\n\n\0"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void encodeFrameWithHeadersThatShouldBeEscaped() {
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.DISCONNECT);
|
||||
headers.addNativeHeader("a:\r\n\\b", "alpha:bravo\r\n\\");
|
||||
|
||||
Message<byte[]> frame = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
|
||||
|
||||
assertEquals("DISCONNECT\na\\c\\r\\n\\\\b:alpha\\cbravo\\r\\n\\\\\n\n\0", new StompCodec().encoder().apply(frame).asString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void encodeFrameWithHeadersBody() {
|
||||
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);
|
||||
headers.addNativeHeader("a", "alpha");
|
||||
|
||||
Message<byte[]> frame = MessageBuilder.withPayloadAndHeaders("Message body".getBytes(), headers).build();
|
||||
|
||||
assertEquals("SEND\na:alpha\ncontent-length:12\n\nMessage body\0", new StompCodec().encoder().apply(frame).asString());
|
||||
}
|
||||
|
||||
private Message<byte[]> decode(String stompFrame) {
|
||||
this.decoder.apply(Buffer.wrap(stompFrame));
|
||||
return consumer.arguments.get(0);
|
||||
}
|
||||
|
||||
private static final class ArgumentCapturingConsumer<T> implements Consumer<T> {
|
||||
|
||||
private final List<T> arguments = new ArrayList<T>();
|
||||
|
||||
@Override
|
||||
public void accept(T t) {
|
||||
arguments.add(t);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -1,153 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2013 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.messaging.simp.stomp;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
|
||||
import org.springframework.messaging.simp.SimpMessageType;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author Rossen Stoyanchev
|
||||
*/
|
||||
public class StompMessageConverterTests {
|
||||
|
||||
private StompMessageConverter converter;
|
||||
|
||||
|
||||
@Before
|
||||
public void setup() {
|
||||
this.converter = new StompMessageConverter();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectFrame() throws Exception {
|
||||
|
||||
String accept = "accept-version:1.1";
|
||||
String host = "host:github.org";
|
||||
|
||||
TextMessage textMessage = StompTextMessageBuilder.create(StompCommand.CONNECT)
|
||||
.headers(accept, host).build();
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<byte[]> message = (Message<byte[]>) this.converter.toMessage(textMessage.getPayload());
|
||||
|
||||
assertEquals(0, message.getPayload().length);
|
||||
|
||||
MessageHeaders headers = message.getHeaders();
|
||||
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
|
||||
Map<String, Object> map = stompHeaders.toMap();
|
||||
assertEquals(5, map.size());
|
||||
assertNotNull(stompHeaders.getId());
|
||||
assertNotNull(stompHeaders.getTimestamp());
|
||||
assertEquals(SimpMessageType.CONNECT, stompHeaders.getMessageType());
|
||||
assertEquals(StompCommand.CONNECT, stompHeaders.getCommand());
|
||||
assertNotNull(map.get(SimpMessageHeaderAccessor.NATIVE_HEADERS));
|
||||
|
||||
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
|
||||
assertEquals("github.org", stompHeaders.getHost());
|
||||
|
||||
assertEquals(SimpMessageType.CONNECT, stompHeaders.getMessageType());
|
||||
assertEquals(StompCommand.CONNECT, stompHeaders.getCommand());
|
||||
assertNotNull(headers.get(MessageHeaders.ID));
|
||||
assertNotNull(headers.get(MessageHeaders.TIMESTAMP));
|
||||
|
||||
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");
|
||||
|
||||
assertEquals("CONNECT\n", convertedBack.substring(0,8));
|
||||
assertTrue(convertedBack.contains(accept));
|
||||
assertTrue(convertedBack.contains(host));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectWithEscapes() throws Exception {
|
||||
|
||||
String accept = "accept-version:1.1";
|
||||
String host = "ho\\c\\ns\\rt:st\\nomp.gi\\cthu\\b.org";
|
||||
|
||||
TextMessage textMessage = StompTextMessageBuilder.create(StompCommand.CONNECT)
|
||||
.headers(accept, host).build();
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<byte[]> message = (Message<byte[]>) this.converter.toMessage(textMessage.getPayload());
|
||||
|
||||
assertEquals(0, message.getPayload().length);
|
||||
|
||||
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
|
||||
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
|
||||
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.toNativeHeaderMap().get("ho:\ns\rt").get(0));
|
||||
|
||||
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");
|
||||
|
||||
assertEquals("CONNECT\n", convertedBack.substring(0,8));
|
||||
assertTrue(convertedBack.contains(accept));
|
||||
assertTrue(convertedBack.contains(host));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectCR12() throws Exception {
|
||||
|
||||
String accept = "accept-version:1.2\n";
|
||||
String host = "host:github.org\n";
|
||||
String test = "CONNECT\r\n" + accept.replaceAll("\n", "\r\n") + host.replaceAll("\n", "\r\n") + "\r\n";
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<byte[]> message = (Message<byte[]>) this.converter.toMessage(test.getBytes("UTF-8"));
|
||||
|
||||
assertEquals(0, message.getPayload().length);
|
||||
|
||||
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
|
||||
assertEquals(Collections.singleton("1.2"), stompHeaders.getAcceptVersion());
|
||||
assertEquals("github.org", stompHeaders.getHost());
|
||||
|
||||
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");
|
||||
|
||||
assertEquals("CONNECT\n", convertedBack.substring(0,8));
|
||||
assertTrue(convertedBack.contains(accept));
|
||||
assertTrue(convertedBack.contains(host));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectWithEscapesAndCR12() throws Exception {
|
||||
|
||||
String accept = "accept-version:1.1\n";
|
||||
String host = "ho\\c\\ns\\rt:st\\nomp.gi\\cthu\\b.org\n";
|
||||
String test = "\n\n\nCONNECT\r\n" + accept.replaceAll("\n", "\r\n") + host.replaceAll("\n", "\r\n") + "\r\n";
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<byte[]> message = (Message<byte[]>) this.converter.toMessage(test.getBytes("UTF-8"));
|
||||
|
||||
assertEquals(0, message.getPayload().length);
|
||||
|
||||
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
|
||||
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
|
||||
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.toNativeHeaderMap().get("ho:\ns\rt").get(0));
|
||||
|
||||
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");
|
||||
|
||||
assertEquals("CONNECT\n", convertedBack.substring(0,8));
|
||||
assertTrue(convertedBack.contains(accept));
|
||||
assertTrue(convertedBack.contains(host));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
package org.springframework.messaging.simp.stomp;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashSet;
|
||||
|
||||
@@ -84,7 +85,7 @@ public class StompProtocolHandlerTests {
|
||||
|
||||
assertEquals(1, this.session.getSentMessages().size());
|
||||
textMessage = (TextMessage) this.session.getSentMessages().get(0);
|
||||
Message<?> message = new StompMessageConverter().toMessage(textMessage.getPayload());
|
||||
Message<?> message = new StompDecoder().decode(ByteBuffer.wrap(textMessage.getPayload().getBytes()));
|
||||
StompHeaderAccessor replyHeaders = StompHeaderAccessor.wrap(message);
|
||||
|
||||
assertEquals(StompCommand.CONNECTED, replyHeaders.getCommand());
|
||||
|
||||
Reference in New Issue
Block a user