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

Use appropriate message id header for STOMP 1.1 vs 1.2

Browse Source 
Issue: SPR-14391
This commit is contained in:
Rossen Stoyanchev
2016-06-27 16:12:59 -04:00
parent 2cdcf752ba
commit ebdc770ed3
1 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompSession.java
View File

@@ -67,7 +67,6 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private static final Message<byte[]> HEARTBEAT; private static final Message<byte[]> HEARTBEAT;
static { static {
StompHeaderAccessor accessor = StompHeaderAccessor.createForHeartbeat(); StompHeaderAccessor accessor = StompHeaderAccessor.createForHeartbeat();
HEARTBEAT = MessageBuilder.createMessage(StompDecoder.HEARTBEAT_PAYLOAD, accessor.getMessageHeaders()); HEARTBEAT = MessageBuilder.createMessage(StompDecoder.HEARTBEAT_PAYLOAD, accessor.getMessageHeaders());
@@ -93,6 +92,8 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private volatile TcpConnection<byte[]> connection; private volatile TcpConnection<byte[]> connection;
private volatile String version;
private final AtomicInteger subscriptionIndex = new AtomicInteger(); private final AtomicInteger subscriptionIndex = new AtomicInteger();
private final Map<String, DefaultSubscription> subscriptions = new ConcurrentHashMap<String, DefaultSubscription>(4); private final Map<String, DefaultSubscription> subscriptions = new ConcurrentHashMap<String, DefaultSubscription>(4);
@@ -313,7 +314,12 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
@Override @Override
public Receiptable acknowledge(String messageId, boolean consumed) { public Receiptable acknowledge(String messageId, boolean consumed) {
StompHeaders stompHeaders = new StompHeaders(); StompHeaders stompHeaders = new StompHeaders();
stompHeaders.setId(messageId); if ("1.1".equals(this.version)) {
stompHeaders.setMessageId(messageId);
}
else {
stompHeaders.setId(messageId);
}
String receiptId = checkOrAddReceipt(stompHeaders); String receiptId = checkOrAddReceipt(stompHeaders);
Receiptable receiptable = new ReceiptHandler(receiptId); Receiptable receiptable = new ReceiptHandler(receiptId);
@@ -407,6 +413,7 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
} }
else if (StompCommand.CONNECTED.equals(command)) { else if (StompCommand.CONNECTED.equals(command)) {
initHeartbeatTasks(stompHeaders); initHeartbeatTasks(stompHeaders);
this.version = stompHeaders.getFirst("version");
this.sessionFuture.set(this); this.sessionFuture.set(this);
this.sessionHandler.afterConnected(this, stompHeaders); this.sessionHandler.afterConnected(this, stompHeaders);
} }