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

Simple broker sends notice after disconnect

Browse Source 
Before this change the simple broker simply removed subscriptions
upon receiving a DISCONNECT message assuming it was a result of
a client STOMP WebSocket session ending.

However, if the server-side application sends a DISCONNECT to
the broker in order to terminate a session, the STOMP WebSocket
session could remain unware without any further action. This
change ensures the simple broker sends a DISCONNECT_ACK message
downstream whenever it receives a DISCONNECT.

Issue: SPR-12288
This commit is contained in:
Rossen Stoyanchev
2014-10-23 21:55:16 -04:00
parent 6463878f2d
commit 01aa64c534
4 changed files with 21 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageType.java
View File

@@ -39,6 +39,8 @@ public enum SimpMessageType {
DISCONNECT,
DISCONNECT_ACK,
OTHER;
}

5
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -169,6 +169,11 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
SimpMessageHeaderAccessor disconnectAck = SimpMessageHeaderAccessor.create(SimpMessageType.DISCONNECT_ACK);
initHeaders(disconnectAck);
disconnectAck.setSessionId(sessionId);
Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, disconnectAck.getMessageHeaders());
getClientOutboundChannel().send(messageOut);
}
else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
if (logger.isDebugEnabled()) {

15
spring-messaging/src/test/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandlerTests.java
View File

@@ -16,6 +16,10 @@
package org.springframework.messaging.simp.broker;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.Collections;
import org.junit.Before;
@@ -24,7 +28,6 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
@@ -32,9 +35,6 @@ import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageBuilder;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
/**
* Unit tests for SimpleBrokerMessageHandler.
*
@@ -115,7 +115,12 @@ public class SimpleBrokerMessageHandlerTests {
this.messageHandler.handleMessage(createMessage("/foo", "message1"));
this.messageHandler.handleMessage(createMessage("/bar", "message2"));
verify(this.clientOutboundChannel, times(3)).send(this.messageCaptor.capture());
verify(this.clientOutboundChannel, times(4)).send(this.messageCaptor.capture());
Message<?> captured = this.messageCaptor.getAllValues().get(0);
assertEquals(SimpMessageType.DISCONNECT_ACK, SimpMessageHeaderAccessor.getMessageType(captured.getHeaders()));
assertEquals(sess1, SimpMessageHeaderAccessor.getSessionId(captured.getHeaders()));
assertCapturedMessage(sess2, "sub1", "/foo");
assertCapturedMessage(sess2, "sub2", "/foo");
assertCapturedMessage(sess2, "sub3", "/bar");

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

@@ -393,6 +393,10 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
if (SimpMessageType.CONNECT_ACK.equals(stompAccessor.getMessageType())) {
stompAccessor = convertConnectAcktoStompConnected(stompAccessor);
}
else if (SimpMessageType.DISCONNECT_ACK.equals(stompAccessor.getMessageType())) {
stompAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
stompAccessor.setMessage("Session closed.");
}
else if (stompAccessor.getCommand() == null || StompCommand.SEND.equals(stompAccessor.getCommand())) {
stompAccessor.updateStompCommandAsServerMessage();
}